從 Spark Streaming 到 Apache Flink : 實時資料流在愛奇藝的演進

作者：陳越晨

整理：劉河

本文将為大家介紹 Apache Flink 在愛奇藝的生産與實踐過程。你可以借此了解到愛奇藝引入Apache Flink的背景與挑戰，以及平台建構化流程。主要内容如下：

1. 愛奇藝在實時計算方面的的演化和遇到的一些挑戰
2. 愛奇藝使用Flink的User Case
3. 愛奇藝Flink平台化建構流程
4. 愛奇藝在Flink上的改進
5. 未來工作

愛奇藝簡介

愛奇藝在2010年正式上線，于2018年3月份在納斯達克上市。我們擁有規模龐大且高度活躍的使用者基礎，月活躍使用者數5.65億人，在線上視訊領域名列第一。在移動端，愛奇藝月度總有效時長59.08億小時，穩居中國APP榜第三名。

一、愛奇藝在實時計算方面的演化和遇到的一些挑戰

1. 實時計算在愛奇藝的演化過程

實時計算是基于一些實時到達、速率不可控、到達次序獨立不保證順序、一經處理無法重放除非特意儲存的無序時間序列的資料的線上計算。

是以，在實時計算中，會遇到資料亂序、資料延時、事件時間與處理時間不一緻等問題。愛奇藝的峰值事件數達到1100萬/秒，在正确性、容錯、性能、延遲、吞吐量、擴充性等方面均遇到不小的挑戰。

愛奇藝從2013年開始小規模使用storm，部署了3個獨立叢集。在2015年，開始引入Spark Streaming，部署在YARN上。在2016年，将Spark Streaming平台化，建構流計算平台，降低使用者使用成本，之後流計算開始在愛奇藝大規模使用。在2017年，因為Spark Streaming的先天缺陷，引入Flink，部署在獨立叢集和YARN上。在2018年，建構Streaming SQL與實時分析平台，進一步降低使用者使用門檻。

2. 從Spark Streaming到

愛奇藝主要使用的是Spark Streaming和Flink來進行流式計算。Spark Streaming的實作非常簡單，通過微批次将實時資料拆成一個個批處理任務，通過批處理的方式完成各個子Batch。Spark Streaming的API也非常簡單靈活，既可以用DStream的java/scala API，也可以使用SQL定義處理邏輯。但Spark Streaming受限于微批次處理模型，業務方需要完成一個真正意義上的實時計算會非常困難，比如基于資料事件時間、資料晚到後的處理，都得使用者進行大量程式設計實作。愛奇藝這邊大量使用Spark Streaming的場景往往都在于實時資料的采集落盤。

架構的實時計算模型是基于Dataflow Model實作的，完全支援Dataflow Model的四個問題：What，支援定義DAG圖；Where：定義各類視窗（固定視窗、滑動視窗和Session視窗）；When：支援靈活定義計算觸發時間；How：支援豐富的Function定義資料更新模式。和Spark Streaming一樣，Flink支援分層API，支援DataStream API，Process Function，SQL。Flink最大特點在于其實時計算的正确性保證：Exactly once，原生支援事件時間，支援延時資料處理。由于Flink本身基于原生資料流計算，可以達到毫秒級低延時。

在愛奇藝實測下來，相比Spark Streaming，Apache Flink在相近的吞吐量上，有更低的延時，更好的實時計算表述能力，原生實時事件時間、延時資料處理等。

二、在愛奇藝使用 Flink 的一些案例

下面通過三個Use Case來介紹一下，愛奇藝具體是怎麼使用Flink的，包括海量資料實時ETL，實時風控，分布式調用鍊分析。

1. 海量資料實時ETL

在愛奇藝這邊所有使用者在端上的任何行為都會發一條日志到nginx伺服器上，總量超過千萬QPS。對于具體某個業務來說，他們後續做實時分析，隻希望通路到業務自身的資料，于是這中間就涉及一個資料拆分的工作。

在引入Flink之前，最早的資料拆分邏輯是這樣子的，在Ngnix機器上通過“tail -f /xxx/ngnix.log | grep "xxx"”的方式，配置了無數條這樣的規則，将這些不同的資料按照不同的規則，打到不同的業務kafka中。但這樣的規則随着業務線的規模的擴大，這個tail程序越來越多，逐漸遇到了伺服器性能瓶頸。

于是，我們就有了這樣一個設想，希望通過實時流計算将資料拆分到各個業務kafka。具體來說，就是Nginx上的全量資料，全量采集到一級Kafka，通過實時ETL程式，按需将資料采集到各個業務Kafka中。當時，愛奇藝主的實時流計算基本均是基于Spark Streaming的，但考慮到Spark Streaming延遲相對來說比較高，愛奇藝從這個case展開開始推進Apache Flink的應用。

海量資料實時ETL的具體實作，主要有以下幾個步驟：

1. 解碼：各個端的投遞日志格式不統一，需要首先将各個端的日志按照各種解碼方式解析成規範化的格式，這邊選用的是JSON
2. 風控：實時拆分這邊的資料都會過一下風控的規則，過濾掉很大一部分刷量日志。由于量級太高，如果将每條日志都過一下風控規則，延時會非常大。這邊做了幾個優化，首先，将使用者資料通過DeviceID拆分，不同的DeviceID拆分到不同的task manager上，每個task manager用本地記憶體做一級緩存，将redis和flink部署在一起，用本地redis做二級緩存。最終的效果是，每秒redis通路降到了平均4k，實時拆分的P99延時小于500ms。
3. 拆分：按照各個業務進行拆分
4. 采樣、再過濾：根據每個業務的拆分過程中根據使用者的需求不同，有采樣、再過濾等過程

2. 實時風控

防機器撞庫盜号攻擊是安全風控的一個常見需求，主要需求集中于事中和事後。在事中，進行超高頻異常檢測分析，過濾使用者異常行為；在事後，生成IP和裝置ID的黑名單，供各業務實時分析時進行防刷使用。

以下是兩個使用

特性的案例：

1. CEP：因為很多黑産使用者是有固定的一些套路，比如剛注冊的使用者可能在短時間内會進行一兩項操作，我們通過CEP模式比對，過濾掉那些有固定套路的黑産行為
2. 多視窗聚合：風控這邊會有一些需求，它需要在不同的一些時間視窗，有些時間視窗要求比較苛刻，可能是需要在一秒内或亞秒内去看一下某個使用者有多少次通路，然後對他進行計數，計數的結果超過某些門檻值就判斷他是異常使用者。通過Flink低延時且支援多視窗的特點，進行超高頻的異常檢測，比如對同一個使用者在1秒内的請求進行計數，超過某個門檻值的話就會被識别成黑産。

3. 分布式追蹤系統

分布式調用鍊追蹤系統，即全鍊路監控，每個公司基本都會有。在一個微服務架構當中，服務間的調用關系錯綜複雜，往往很難排查問題，識别性能性能瓶頸，這時候就需要分布式調用鍊追蹤系統了。

上圖是一個調用鍊的追蹤拓撲圖，每個點是一個具體的一個應用，就是具體經過哪個應用，每條邊是說明這個應用到下一個應用當中耗時了多久。

除了宏觀分析外，業務還想去看具體某一條日志的分析，具體某一次調用它是哪裡慢了，哪裡快了？是以，調用鍊還有另外一個需求，就是對于具體某次調用，想看一下它的具體耗時。

系統簡單架構如上圖，上半部分偏重于埋點，下半部分偏于分析。埋點簡單來講，就是通過用戶端SDK埋點以及Agent采集，将系統調用日志全部打到Kafka中，我們通過Flink對他們進行各類分析。對于統計類的分析，就是通過Flink計算存儲到HBase當中，提供一些監控報警、調用鍊拓普查詢等這種分析。針對這類需求，我們運用了Flink的多視窗聚合的特性，通過一分鐘或者多分鐘的視窗，從茫茫日志中尋找哪條是實際的調用鍊，建構APP各個應用的拓撲調用關系，第二級是基于第一級分析的一個結果，分析出那個拓普圖按各個視窗、各個不同的邊去算每條邊的平均耗時的統計。除此之外，我們還将通過Flink将原始資料打到ES裡面供使用者直接去查詢。

三、Flink平台化

1. 概覽

接下來将主要介紹愛奇藝的大資料平台的建構。上圖不限于Flink，是大資料平台的整體架構圖。在愛奇藝，存儲層基本是基于Hadoop生态的，比如像HDFS、HBase、Kudu等；計算層，使用YARN，支援MapReduce、Spark、

、Hive、Impala等這些引擎；資料開發層，主要是一些自研産品，批處理開發在愛奇藝有工作流開發，資料內建等。實時計算開發，有流計算開發、Streaming SQL、實時分析等平台工具可以使用。

接下來，我們将簡單介紹愛奇藝實時計算與分析平台。

2. 實時計算平台

2.1 流任務平台

流任務平台是愛奇藝實時計算的底層平台，支援流任務的送出運作與管理。流任務平台支援YARN, Mesos, Flink獨立叢集等多種資源排程架構；支援Storm, Spark Streaming,

, Streaming SQL等計算任務的托管與運作。在功能上，我們支援使用者直接打包程式上傳部署流任務，也支援使用者通過Streaming SQL工具編寫SQL進行流計算開發。為了更好地對計算任務進行管理，流計算平台提供JAR包、函數管理，任務名額監控，以及資源審計功能。

2.2 Streaming SQL

無論對于Spark Streaming還是Flink來說，他們均有一個較好的SQL優化引擎，但均缺乏DDL、DML建立的語義。于是對于業務來說，均需要業務先程式設計定義Source以及Sink，才可以使用SQL進行後續開發。

是以，愛奇藝自研的Streaming SQL定義了一套DDL和DML文法。其中，我們定義了4種表：

流表：定義了輸入源是什麼？具體的解碼方式是什麼？系統支援Json的解碼方式，也支援使用者自定義解碼函數。

次元表：主要是靜态表，支援MySQL，主要是用于流表Join的。

臨時表：和Hive的臨時表類似，使用者定義中間過程。

結果表：定義了具體輸出的類型，輸出的源是什麼？怎麼通路？這邊的輸出源支援，就是常見的比如Kafka、MySQL、Kudu、ES、Druid、HBase等這樣一些分析型資料庫。

為了更好地支援業務需求，StreamingSQL預設也支援IP庫相關的預定義函數，也支援使用者自定義函數。

上圖是一個StreamingSQL的應用Case，将P99，P50耗時列印到Console中。

為了更好地支援業務使用Streaming SQL，StreamingSQL提供Web IDE，提供代碼高亮、關鍵詞提示、文法檢查、代碼調試等功能。

3. 實時分析平台

實時分析平台，是愛奇藝基于Druid建構的分鐘級延時的實時分析平台，支援通過Web向導配置，完成超大規模實時資料多元度的分析，并生成分鐘級延時的可視化報表。支援的功能有，接入實時資料進行OLAP分析；制作實時報警；生産實時資料接口，配置監控報警等。

産品優勢：

• 全向導配置：從實時資料到報表生成僅需向導配置即可
• 計算存儲透明：無需管理大資料處理任務與資料存儲
• 分鐘級低延時: 從資料産生到報表展示隻有1分鐘延時
• 秒級查詢：亞秒級傳回分析報表
• 支援靈活變更需求：業務可靈活更改次元，重新上線即可生效

3.1 使用者向導配置

實時分析平台，将整個分析流程抽象成資料接入，資料處理，模型配置和報表配置4個過程。其中，模型配置完全按照OLAP模型，要求實時資料符合星型模型，存在時間戳、名額、次元等字段。

3.2 資料處理配置

在資料處理層，實時分析平台提供向導配置頁面，支援使用者通過純頁面的方式就可以配置資料處理過程，這主要應對一些簡單場景，針對部分連SQL都不熟悉的小白使用者提供頁面配置方案；初次之外，類似StreamingSQL，實時分析也提供使用者自定義SQL方式定義資料處理過程。

四、Flink改進

在Flink平台化的時候，我們遇到了幾個

的問題，分别對其進行了些改進。

1. 改進 - 優雅恢複checkpoint

第一個改進是關于checkpoint的優雅恢複。這個問題的出發點是，業務希望使用Spark Streaming可以通過代碼控制從哪個checkpoint恢複，但對于Flink來講，業務沒法通過代碼控制checkpoint恢複點，需要手動指定檢查點去恢複checkpoint。于是，我們希望Flink可以像Spark Streaming一樣，直接通過代碼方式恢複checkpoint。

針對這個問題，我們修改源碼，在Flink任務啟動時，從實際的路徑當中找到他最新的一個checkpoint，直接從那個checkpoint當中恢複，當然這個也是可以讓使用者選的，他如果還想用原生方式恢複也可以，但提供一個選項，它可以支援從最近的checkpoint恢複。

2. 改進 - Kafka Broker HA

第二個改進是關于Kafka Broker HA的一個問題，比如像Kafka Broker故障的時候，Kafka還可以正常工作，但Flink程式往往會挂掉。針對這個問題，我們處理了Flink在Kafka Broker退出之後的sockerTimeOutException，支援使用者重試次數配置來解決這個問題。

五、Flink未來工作

最後，介紹一下愛奇藝在

Apache Flink的

未來工作。目前StreamingSQL還隻支援Spark Streaming和Structured Streaming引擎，後續很快會支援Flink引擎，大幅降低業務的Flink開發成本。随着Flink任務規模不斷變大，我們将重點提升Flink在愛奇藝的成熟度，完善監控報警，增加資源審計流程（目前還僅對Spark Streaming進行資源審計）。另外，我們要研究下Flink 1.6的一些新特性，嘗試下Kafka 2.0，調研Exactly once方案；另外，我們将對Flink新版本進行一些嘗試，推進批流統一。