摘要：本文由愛奇藝大資料服務負責人梁建煌分享，介紹愛奇藝如何基于 Apache Flink 技術打造實時計算平台，并通過業務應用案例分享幫助使用者了解 Apache Flink 的技術特點及應用場景。提綱如下：
- 愛奇藝 Flink 服務現狀
- Flink 改進
- 實時計算平台
- Flink 業務案例
- 挑戰與規劃
1.愛奇藝 Flink 服務現狀
愛奇藝從 2012 年開始開展大資料業務，一開始隻有二十幾個節點，主要是 MapReduce、Hive 等離線計算任務。到 2014 年左右上線了 Storm、Spark 實時計算服務，并随後釋出了基于 Spark 的實時計算平台 Europa。2017 年開始引入 Flink，用來替代部分 Spark Streaming 場景，滿足更低延遲的實時計算需求。在這之後，相繼推出流式 SQL 引擎、實時分析平台、實時資料生産平台等一系列工具，用來提升實時計算開發效率。
目前公司内 Flink 類型節點機器 15000 多台，主要有兩種部署模式：
- 混部模式：Flink、Spark、MapReduce 等服務混合部署，15000 多台規模
- 獨立模式：Flink 服務獨立部署，用于重要業務，約 700 多台規模
Flink 作業規模達到 800 個，每日資料生産量維持在萬億級别，日均 2500 TB。
下圖所示為愛奇藝實時計算服務體系：
2.Flink 改進
2.1 監控和報警
Flink 原有的監控比較簡單，無法滿足業務細粒度的監控報警需求。當計算過程出現問題時，無法清晰了解計算作業内部情況，不利于進一步分析。是以，我們改進了 Flink 監控報警機制，增加了很多細粒度的監控名額，主要包括三種：
- Job 級别監控名額：監控 Job 狀态、Checkpoint 狀态及耗時，當 Job 異常時自動通過實時計算平台重新開機。
- Operator 級别監控名額：監控 Flink 任務的時延、反壓、Source/Sink 流量，并對每個 Operator 進行名額聚合，以便使用者檢視。
- TaskManager 級别監控名額：監控 CPU 使用率、記憶體使用率、JVM GC 等正常名額。
2.2 狀态管理
由于 checkpoint 是 Flink job 内部狀态，當 job 重新開機時，上一個 job 的狀态就丢失掉，導緻部分資料丢失，影響到業務。
針對上述問題，我們對 Flink 作業狀态管理進行了改進。使用者送出 Flink job 時，會在實時計算管理平台上配置 checkpoint 路徑。通過實時計算管理平台重新開機 Flink job 時，先找到上一次成功的 checkpoint，從中恢複 job 丢失的狀态（flink run -s :checkpointPath/chk-n/_metadata）。
改進後解決了狀态丢失的問題，但帶來新的缺陷。對于狀态資料很大的作業，使用 RocksDBStateBackend 做增量 checkpoint，重新開機後，上一個 job 的 checkpoint 被依賴而無法删除。随着 Flink 作業長時間運作且發生多次 job 重新開機，系統中堆積大量無用的 checkpoint。
針對該問題，我們使用 savepoint 方式打斷增量 checkpoint 的依賴鍊：
- 主動重新開機：通過計算平台主動重新開機 Flink job 前，系統會先對 job 進行 savepoint 操作再關閉 job，然後從該 savepoint 啟動（flink run -s :savepointPath）。
- 異常重新開機：當平台監測到 Flink job 異常時，會自動從上次 checkpoint 開始啟動該 job。一旦 job 進入到 RUNNING 狀态，會先做一次 savepoint，解除對上一個 checkpoint 的依賴。
2.3 StreamingSQL
為了便于使用者開發流任務，愛奇藝自研了支援 Spark、Flink 的流式 SQL 引擎 StreamingSQL。使用者隻需要通過編寫 SQL 即可完成流計算 ETL 任務的開發。同時，我們也提供 IDE 編輯器和大量常用的預定義函數。
StreamingSQL 定義了 4 種類型資料表：
- 流表：定義計算邏輯的輸入，目前支援Kafka
- 次元表：靜态表，用于與流表join，比如字典映射
- 臨時表：定義中間結果，簡化子查詢邏輯
- 結果表：定義計算邏輯的輸出
資料從流表流入，通過一系列 SQL 語句描述的計算，計算結果寫入結果表。對于計算邏輯比較複雜的計算，可能需要定義多層嵌套的子查詢對計算邏輯進行描述，此時可以通過定義臨時表，将計算邏輯進行拆分，降低子查詢嵌套的深度。
下圖展示了 StreamingSQL 例子：
3.實時計算平台
愛奇藝從 2015 年開始陸續推出實時計算管理、實時資料生産、實時資料分析等多個平台，滿足作業開發、資料生産、資料分析等不同場景下的開發需求，提升使用者的使用體驗和開發效率。
3.1 實時計算管理平台
實時計算管理平台用于 Spark、Flink 任務的開發與管理。使用者可以在 Web IDE 上配置相關參數進行任務的開發、上傳、啟動、停止等正常操作。計算管理平台提供了大量管理子產品以提高使用者的操作體驗，主要包括以下幾項：
- 檔案管理：通過平台的檔案管理功能使用者可以友善的管理任務的 Jar 包及依賴庫。
- 函數管理：為使用者提供了豐富的系統函數，并支援使用者注冊 UDF。
- 版本管理：使用者可以實作任務、檔案的版本對比及舊版本的復原。
- 系統同時提供了監控大盤、報警訂閱、資源審計、異常診斷等多種功能輔助使用者實時掌握作業情況。
3.2 實時資料處理平台
愛奇藝的資料處理平台經曆了 3 個階段的疊代更新，從原先的離線資料采集系統一步步演變成支撐千萬 QPS 的實時資料生産平台。
■ Venus 1.0 – 資料采集系統
2015 年開始，我們推出了第一代資料采集平台 Venus 1.0。資料來源于兩個方面，從用戶端端收集到的使用者觀看視訊的行為資料及背景服務的日志資料。使用者資料從 PC、App 等用戶端采集投遞給平台後端的 Nginx 接收器，并落盤到本地檔案中，再由 Venus agent 解析檔案進行資料采集。服務日志資料是由機器上的 Venus agent 解析 log 檔案采集。Venus 采集的資料直接上傳到 HDFS 進行後續的離線 ETL 處理，生成離線報表供資料分析使用。
Venus 1.0 版本主要基于 Apache Flume 架構進行開發，并通過 tail+grep、awk、sed 等腳本進行資料過濾。在資料量較小時，該平台很好的解決了資料處理的需求。
■ Venus 2.0 – 實時資料處理平台
在 2017 年，随着資料量的增長及實時業務需求的出現，Venus 1.0 漸漸變得力不從心。衆多業務需求導緻 agent 上存在大量過濾規則，過多占用機器資源甚至影響到機器上服務的穩定性。同時，每次變更都需要重新開機所有 agents，大大提高上線成本及風險。
是以，我們設計實作了實時資料處理平台 Venus 2.0 版本，将實時過濾功能從 Venus agent 遷移到 Flink 中并采用兩級 Kafka 結構。改進後的資料平台無需重新開機即可動态增減資料處理規則，資料處理能力也提升了 10 倍以上，大大優化了平台的實時效果。
■ Venus 3.0 – 實時資料生産平台
随着實時業務的大量增加，Venus 2.0 也帶來了 Kafka 資料備援、不友善分享等問題，我們在 2019 年進行了第三次改造，從資料處理更新到資料生産，推出了實時資料生産平台 Venus 3.0 版本。
使用者可以在新平台上配置實時資料處理規則，并可自由組合 Filter、Split、Window 等常見算子，生産出來的流資料可以存儲到流式數倉裡。流式數倉是我們參考離線數倉概念打造的基于 Kafka 的資料倉庫，用于以資料倉庫的形式統一管理流資料。
借助實時資料生産平台及流式數倉，使用者可以更加便捷地加工實時流資料，并通過業務線間的資料分享來減少流資料的重複生産。
3.3 實時資料分析平台
RAP（Realtime Analysis Platform）是愛奇藝基于 Apache Druid + Spark / Flink 建構的分鐘級延時的實時分析平台，支援通過 web 向導配置完成超大規模實時資料的多元度分析，為使用者提供一體化的 OLAP 分析操作流程，隻需要幾步簡單的配置，即可自動建立 OLAP 模型、生成分鐘級延時的可視化報表，并提供實時報警功能。
RAP 實時分析平台解決了使用者在資料分析中遇到的幾個困難：
1.OLAP 選型困難：愛奇藝目前提供了 Kylin、Impala、Kudu、Druid、ElasticSearch 等不同的資料存儲/查詢引擎，使用者需要了解不同 OLAP 引擎的優缺點，花費大量精力學習，依然可能選錯。RAP 幫使用者屏蔽了這層，無需考慮中間資料、結果資料存到哪裡、怎麼查詢。
2. 開發成本高：使用者需要寫 Spark 或 Flink 代碼進行實時流資料處理，并進行報表前端開發，流程冗長而複雜。在 RAP 實時分析平台上，使用者無需編寫Spark/Flink 程式或 SQL，隻需要通過 web 配置處理規則、分析規則、報表模闆、報警規則即可，大幅降低開發門檻，提升了開發效率，從以往的幾天開發一張報表縮短到半小時。
3. 資料實時性差：從資料産生到資料可被查詢，中間存在較高時延（從數十分鐘到天級别不等），且查詢較慢。借助于 Flink 的實時處理能力，RAP 實作了端到端分鐘級低延時的實時報表功能，且支援大規模資料亞秒級查詢。
- 維護耗費時間：資料源發生改變時，修改的範圍會覆寫整個流程，從資料處理到報表配置全部需要變更，很難操作和維護。RAP 提供了自動更新功能，幫助使用者免去人工維護的麻煩。
RAP 實時分析平台架構圖：
4.Flink 業務案例
4.1 資訊流推薦實時化
愛奇藝很早就開始了基于網格式的長視訊推薦業務，近幾年随着短視訊的興起，資訊流形式的推薦發展迅速。資訊流場景裡，需要在幾秒内根據使用者的觀看行為實時推薦相關性更高的視訊，對資料的時效性要求更高。
原本基于 Spark Streaming 的實時資料處理架構無法滿足這類低延遲的需求，是以，我們協助業務遷移到 Flink 平台上，消除了批量資料處理帶來的延遲。單個任務的延遲從 1 分鐘縮短到 1-2 秒，端到端的性能提升了 86 倍，顯著提升了推薦效果。
4.2 使用 Flink 生産深度學習訓練資料
深度學習大量應用于愛奇藝内部的各項業務，幫助業務更好的挖掘資料的價值。在深度學習場景中，訓練資料的時效性非常關鍵。我們使用 Flink 幫助業務更加實時地生産訓練資料。
下圖所示為愛奇藝廣告點選率預測訓練的架構，業務原先通過 Hive/Spark 離線 ETL 方式生成訓練資料，每 6 小時才能更新一次算法模型，導緻使用者特征關聯不及時、不精确，影響到廣告投放效果。
我們基于 Flink 進行了實時化改造，将最近 24 小時的使用者資料實時寫到 Kafka 中，通過 Flink 與存儲在 HBase 中的過去 7 天的使用者特征進行實時 join，實時産出包含最新使用者特征的訓練資料，将算法模型更新周期縮短到 1 小時以内，進而支援更加實時、精确的 CTR （Click-Through-Rate）預估，大幅提升廣告投放效果。
4.3 端到端 Exactly-Once 處理
當 Kafka 節點出現故障重新開機或進行人工運維時，Flink 作業會重複消費資料導緻資料失準，影響後續的資料處理，比如模型訓練。針對該問題，我們設計實作了基于 Kafka Exactly Once Semantics 及 Flink two-phase commit 特性的端到端 Exactly-Once 處理方案。經過我們測試，該方案會帶來 20% 的計算性能損耗，但資料重複率會從原先的最高 300% 降低到 0，很好地解決了節點重新開機帶來的資料精确度問題。
關于 Exactly-once two-phase commit 的原理，可以閱讀 Apache Flink Blog 上的詳細介紹：
https://flink.apache.org/features/2018/03/01/end-to-end-exactly-once-apache-flink.html5.挑戰與規劃
随着 Flink 在愛奇藝得到越來越廣泛的應用，我們在資源管理、穩定性、實時開發等層面面臨新的挑戰。
接下來，我們會推進流批一體化，進一步完善和推廣 StreamingSQL 技術，降低開發門檻。同時，積極嘗試基于 Flink 的機器學習、Flink on Kubernetes、Flink 動态資源調整等前沿方向。
作者介紹：
梁建煌，愛奇藝大資料服務負責人，2012-碩士畢業于上海交通大學後，先後在 SAP、愛奇藝工作，從 2013 年起開始負責愛奇藝大資料服務體系的建設工作，包括大資料存儲、計算、OLAP 以及開發平台等。