摘要：在移動(dòng)端，愛奇藝月度總有效時(shí)長億小時(shí)，穩(wěn)居中國榜第三名。愛奇藝的峰值事件數(shù)達(dá)到萬秒，在正確性容錯(cuò)性能延遲吞吐量擴(kuò)展性等方面均遇到不小的挑戰(zhàn)。從到愛奇藝主要使用的是和來進(jìn)行流式計(jì)算。

作者：陳越晨

整理：劉河

本文將為大家介紹Apache Flink在愛奇藝的生產(chǎn)與實(shí)踐過程。你可以借此了解到愛奇藝引入Apache Flink的背景與挑戰(zhàn)，以及平臺(tái)構(gòu)建化流程。主要內(nèi)容如下：

愛奇藝在實(shí)時(shí)計(jì)算方面的的演化和遇到的一些挑戰(zhàn)
愛奇藝使用Flink的User Case
愛奇藝Flink平臺(tái)化構(gòu)建流程
愛奇藝在Flink上的改進(jìn)
未來工作

愛奇藝在2010年正式上線，于2018年3月份在納斯達(dá)克上市。我們擁有規(guī)模龐大且高度活躍的用戶基礎(chǔ)，月活躍用戶數(shù)5.65億人，在在線視頻領(lǐng)域名列第一。在移動(dòng)端，愛奇藝月度總有效時(shí)長59.08億小時(shí)，穩(wěn)居中國APP榜第三名。

實(shí)時(shí)計(jì)算是基于一些實(shí)時(shí)到達(dá)、速率不可控、到達(dá)次序獨(dú)立不保證順序、一經(jīng)處理無法重放除非特意保存的無序時(shí)間序列的數(shù)據(jù)的在線計(jì)算。

因此，在實(shí)時(shí)計(jì)算中，會(huì)遇到數(shù)據(jù)亂序、數(shù)據(jù)延時(shí)、事件時(shí)間與處理時(shí)間不一致等問題。愛奇藝的峰值事件數(shù)達(dá)到1100萬/秒，在正確性、容錯(cuò)、性能、延遲、吞吐量、擴(kuò)展性等方面均遇到不小的挑戰(zhàn)。

愛奇藝從2013年開始小規(guī)模使用storm，部署了3個(gè)獨(dú)立集群。在2015年，開始引入Spark Streaming，部署在YARN上。在2016年，將Spark Streaming平臺(tái)化，構(gòu)建流計(jì)算平臺(tái)，降低用戶使用成本，之后流計(jì)算開始在愛奇藝大規(guī)模使用。在2017年，因?yàn)镾park Streaming的先天缺陷，引入Flink，部署在獨(dú)立集群和YARN上。在2018年，構(gòu)建Streaming SQL與實(shí)時(shí)分析平臺(tái)，進(jìn)一步降低用戶使用門檻。

愛奇藝主要使用的是Spark Streaming和Flink來進(jìn)行流式計(jì)算。Spark Streaming的實(shí)現(xiàn)非常簡單，通過微批次將實(shí)時(shí)數(shù)據(jù)拆成一個(gè)個(gè)批處理任務(wù)，通過批處理的方式完成各個(gè)子Batch。Spark Streaming的API也非常簡單靈活，既可以用DStream的java/scala API，也可以使用SQL定義處理邏輯。但Spark Streaming受限于微批次處理模型，業(yè)務(wù)方需要完成一個(gè)真正意義上的實(shí)時(shí)計(jì)算會(huì)非常困難，比如基于數(shù)據(jù)事件時(shí)間、數(shù)據(jù)晚到后的處理，都得用戶進(jìn)行大量編程實(shí)現(xiàn)。愛奇藝這邊大量使用Spark Streaming的場(chǎng)景往往都在于實(shí)時(shí)數(shù)據(jù)的采集落盤。

Apache Flink框架的實(shí)時(shí)計(jì)算模型是基于Dataflow Model實(shí)現(xiàn)的，完全支持Dataflow Model的四個(gè)問題：What，支持定義DAG圖；Where：定義各類窗口（固定窗口、滑動(dòng)窗口和Session窗口）；When：支持靈活定義計(jì)算觸發(fā)時(shí)間；How：支持豐富的Function定義數(shù)據(jù)更新模式。和Spark Streaming一樣，F(xiàn)link支持分層API，支持DataStream API，Process Function，SQL。Flink最大特點(diǎn)在于其實(shí)時(shí)計(jì)算的正確性保證：Exactly once，原生支持事件時(shí)間，支持延時(shí)數(shù)據(jù)處理。由于Flink本身基于原生數(shù)據(jù)流計(jì)算，可以達(dá)到毫秒級(jí)低延時(shí)。

在愛奇藝實(shí)測(cè)下來，相比Spark Streaming，Apache Flink在相近的吞吐量上，有更低的延時(shí)，更好的實(shí)時(shí)計(jì)算表述能力，原生實(shí)時(shí)事件時(shí)間、延時(shí)數(shù)據(jù)處理等。

下面通過三個(gè)Use Case來介紹一下，愛奇藝具體是怎么使用Flink的，包括海量數(shù)據(jù)實(shí)時(shí)ETL，實(shí)時(shí)風(fēng)控，分布式調(diào)用鏈分析。

在愛奇藝這邊所有用戶在端上的任何行為都會(huì)發(fā)一條日志到nginx服務(wù)器上，總量超過千萬QPS。對(duì)于具體某個(gè)業(yè)務(wù)來說，他們后續(xù)做實(shí)時(shí)分析，只希望訪問到業(yè)務(wù)自身的數(shù)據(jù)，于是這中間就涉及一個(gè)數(shù)據(jù)拆分的工作。

在引入Flink之前，最早的數(shù)據(jù)拆分邏輯是這樣子的，在Ngnix機(jī)器上通過“tail -f /xxx/ngnix.log | grep "xxx"”的方式，配置了無數(shù)條這樣的規(guī)則，將這些不同的數(shù)據(jù)按照不同的規(guī)則，打到不同的業(yè)務(wù)kafka中。但這樣的規(guī)則隨著業(yè)務(wù)線的規(guī)模的擴(kuò)大，這個(gè)tail進(jìn)程越來越多，逐漸遇到了服務(wù)器性能瓶頸。

于是，我們就有了這樣一個(gè)設(shè)想，希望通過實(shí)時(shí)流計(jì)算將數(shù)據(jù)拆分到各個(gè)業(yè)務(wù)kafka。具體來說，就是Nginx上的全量數(shù)據(jù)，全量采集到一級(jí)Kafka，通過實(shí)時(shí)ETL程序，按需將數(shù)據(jù)采集到各個(gè)業(yè)務(wù)Kafka中。當(dāng)時(shí)，愛奇藝主的實(shí)時(shí)流計(jì)算基本均是基于Spark Streaming的，但考慮到Spark Streaming延遲相對(duì)來說比較高，愛奇藝從這個(gè)case展開開始推進(jìn)Apache Flink的應(yīng)用。

海量數(shù)據(jù)實(shí)時(shí)ETL的具體實(shí)現(xiàn)，主要有以下幾個(gè)步驟：

解碼：各個(gè)端的投遞日志格式不統(tǒng)一，需要首先將各個(gè)端的日志按照各種解碼方式解析成規(guī)范化的格式，這邊選用的是JSON

風(fēng)控：實(shí)時(shí)拆分這邊的數(shù)據(jù)都會(huì)過一下風(fēng)控的規(guī)則，過濾掉很大一部分刷量日志。由于量級(jí)太高，如果將每條日志都過一下風(fēng)控規(guī)則，延時(shí)會(huì)非常大。這邊做了幾個(gè)優(yōu)化，首先，將用戶數(shù)據(jù)通過DeviceID拆分，不同的DeviceID拆分到不同的task manager上，每個(gè)task manager用本地內(nèi)存做一級(jí)緩存，將redis和flink部署在一起，用本地redis做二級(jí)緩存。最終的效果是，每秒redis訪問降到了平均4k，實(shí)時(shí)拆分的P99延時(shí)小于500ms。

拆分：按照各個(gè)業(yè)務(wù)進(jìn)行拆分

采樣、再過濾：根據(jù)每個(gè)業(yè)務(wù)的拆分過程中根據(jù)用戶的需求不同，有采樣、再過濾等過程

2. 實(shí)時(shí)風(fēng)控

防機(jī)器撞庫盜號(hào)攻擊是安全風(fēng)控的一個(gè)常見需求，主要需求集中于事中和事后。在事中，進(jìn)行超高頻異常檢測(cè)分析，過濾用戶異常行為；在事后，生成IP和設(shè)備ID的黑名單，供各業(yè)務(wù)實(shí)時(shí)分析時(shí)進(jìn)行防刷使用。

以下是兩個(gè)使用Flink特性的案例：

CEP：因?yàn)楹芏嗪诋a(chǎn)用戶是有固定的一些套路，比如剛注冊(cè)的用戶可能在短時(shí)間內(nèi)會(huì)進(jìn)行一兩項(xiàng)操作，我們通過CEP模式匹配，過濾掉那些有固定套路的黑產(chǎn)行為

多窗口聚合：風(fēng)控這邊會(huì)有一些需求，它需要在不同的一些時(shí)間窗口，有些時(shí)間窗口要求比較苛刻，可能是需要在一秒內(nèi)或亞秒內(nèi)去看一下某個(gè)用戶有多少次訪問，然后對(duì)他進(jìn)行計(jì)數(shù)，計(jì)數(shù)的結(jié)果超過某些閾值就判斷他是異常用戶。通過Flink低延時(shí)且支持多窗口的特點(diǎn)，進(jìn)行超高頻的異常檢測(cè)，比如對(duì)同一個(gè)用戶在1秒內(nèi)的請(qǐng)求進(jìn)行計(jì)數(shù)，超過某個(gè)閾值的話就會(huì)被識(shí)別成黑產(chǎn)。

分布式調(diào)用鏈追蹤系統(tǒng)，即全鏈路監(jiān)控，每個(gè)公司基本都會(huì)有。在一個(gè)微服務(wù)架構(gòu)當(dāng)中，服務(wù)間的調(diào)用關(guān)系錯(cuò)綜復(fù)雜，往往很難排查問題，識(shí)別性能性能瓶頸，這時(shí)候就需要分布式調(diào)用鏈追蹤系統(tǒng)了。

上圖是一個(gè)調(diào)用鏈的追蹤拓?fù)鋱D，每個(gè)點(diǎn)是一個(gè)具體的一個(gè)應(yīng)用，就是具體經(jīng)過哪個(gè)應(yīng)用，每條邊是說明這個(gè)應(yīng)用到下一個(gè)應(yīng)用當(dāng)中耗時(shí)了多久。

除了宏觀分析外，業(yè)務(wù)還想去看具體某一條日志的分析，具體某一次調(diào)用它是哪里慢了，哪里快了？所以，調(diào)用鏈還有另外一個(gè)需求，就是對(duì)于具體某次調(diào)用，想看一下它的具體耗時(shí)。

系統(tǒng)簡單架構(gòu)如上圖，上半部分偏重于埋點(diǎn)，下半部分偏于分析。埋點(diǎn)簡單來講，就是通過客戶端SDK埋點(diǎn)以及Agent采集，將系統(tǒng)調(diào)用日志全部打到Kafka中，我們通過Flink對(duì)他們進(jìn)行各類分析。對(duì)于統(tǒng)計(jì)類的分析，就是通過Flink計(jì)算存儲(chǔ)到HBase當(dāng)中，提供一些監(jiān)控報(bào)警、調(diào)用鏈拓普查詢等這種分析。針對(duì)這類需求，我們運(yùn)用了Flink的多窗口聚合的特性，通過一分鐘或者多分鐘的窗口，從茫茫日志中尋找哪條是實(shí)際的調(diào)用鏈，構(gòu)建APP各個(gè)應(yīng)用的拓?fù)湔{(diào)用關(guān)系，第二級(jí)是基于第一級(jí)分析的一個(gè)結(jié)果，分析出那個(gè)拓普?qǐng)D按各個(gè)窗口、各個(gè)不同的邊去算每條邊的平均耗時(shí)的統(tǒng)計(jì)。除此之外，我們還將通過Flink將原始數(shù)據(jù)打到ES里面供用戶直接去查詢。

接下來將主要介紹愛奇藝的大數(shù)據(jù)平臺(tái)的構(gòu)建。上圖不限于Flink，是大數(shù)據(jù)平臺(tái)的整體架構(gòu)圖。在愛奇藝，存儲(chǔ)層基本是基于Hadoop生態(tài)的，比如像HDFS、HBase、Kudu等；計(jì)算層，使用YARN，支持MapReduce、Spark、Flink、Hive、Impala等這些引擎；數(shù)據(jù)開發(fā)層，主要是一些自研產(chǎn)品，批處理開發(fā)在愛奇藝有工作流開發(fā)，數(shù)據(jù)集成等。實(shí)時(shí)計(jì)算開發(fā)，有流計(jì)算開發(fā)、Streaming SQL、實(shí)時(shí)分析等平臺(tái)工具可以使用。

接下來，我們將簡單介紹愛奇藝實(shí)時(shí)計(jì)算與分析平臺(tái)。

2.1 流任務(wù)平臺(tái)

流任務(wù)平臺(tái)是愛奇藝實(shí)時(shí)計(jì)算的底層平臺(tái)，支持流任務(wù)的提交運(yùn)行與管理。流任務(wù)平臺(tái)支持YARN, Mesos, Flink獨(dú)立集群等多種資源調(diào)度框架；支持Storm, Spark Streaming, Flink, Streaming SQL等計(jì)算任務(wù)的托管與運(yùn)行。在功能上，我們支持用戶直接打包程序上傳部署流任務(wù)，也支持用戶通過Streaming SQL工具編寫SQL進(jìn)行流計(jì)算開發(fā)。為了更好地對(duì)計(jì)算任務(wù)進(jìn)行管理，流計(jì)算平臺(tái)提供JAR包、函數(shù)管理，任務(wù)指標(biāo)監(jiān)控，以及資源審計(jì)功能。

2.2 Streaming SQL

無論對(duì)于Spark Streaming還是Flink來說，他們均有一個(gè)較好的SQL優(yōu)化引擎，但均缺乏DDL、DML創(chuàng)建的語義。于是對(duì)于業(yè)務(wù)來說，均需要業(yè)務(wù)先編程定義Source以及Sink，才可以使用SQL進(jìn)行后續(xù)開發(fā)。

因此，愛奇藝自研的Streaming SQL定義了一套DDL和DML語法。其中，我們定義了4種表： 流表：定義了輸入源是什么？具體的解碼方式是什么？系統(tǒng)支持Json的解碼方式，也支持用戶自定義解碼函數(shù)。 維度表：主要是靜態(tài)表，支持MySQL，主要是用于流表Join的。 臨時(shí)表：和Hive的臨時(shí)表類似，用戶定義中間過程。 結(jié)果表：定義了具體輸出的類型，輸出的源是什么？怎么訪問？這邊的輸出源支持，就是常見的比如Kafka、MySQL、Kudu、ES、Druid、HBase等這樣一些分析型數(shù)據(jù)庫。

為了更好地支持業(yè)務(wù)需求，StreamingSQL默認(rèn)也支持IP庫相關(guān)的預(yù)定義函數(shù)，也支持用戶自定義函數(shù)。

上圖是一個(gè)StreamingSQL的應(yīng)用Case，將P99，P50耗時(shí)打印到Console中。

為了更好地支持業(yè)務(wù)使用Streaming SQL，StreamingSQL提供Web IDE，提供代碼高亮、關(guān)鍵詞提示、語法檢查、代碼調(diào)試等功能。

實(shí)時(shí)分析平臺(tái)，是愛奇藝基于Druid構(gòu)建的分鐘級(jí)延時(shí)的實(shí)時(shí)分析平臺(tái)，支持通過Web向?qū)渲茫瓿沙笠?guī)模實(shí)時(shí)數(shù)據(jù)多維度的分析，并生成分鐘級(jí)延時(shí)的可視化報(bào)表。支持的功能有，接入實(shí)時(shí)數(shù)據(jù)進(jìn)行OLAP分析；制作實(shí)時(shí)報(bào)警；生產(chǎn)實(shí)時(shí)數(shù)據(jù)接口，配置監(jiān)控報(bào)警等。

產(chǎn)品優(yōu)勢(shì)：

全向?qū)渲茫簭膶?shí)時(shí)數(shù)據(jù)到報(bào)表生成僅需向?qū)渲眉纯?/p>

計(jì)算存儲(chǔ)透明：無需管理大數(shù)據(jù)處理任務(wù)與數(shù)據(jù)存儲(chǔ)

分鐘級(jí)低延時(shí): 從數(shù)據(jù)產(chǎn)生到報(bào)表展示只有1分鐘延時(shí)

秒級(jí)查詢：亞秒級(jí)返回分析報(bào)表

支持靈活變更需求：業(yè)務(wù)可靈活更改維度，重新上線即可生效

3.1 用戶向?qū)渲?/h4>

實(shí)時(shí)分析平臺(tái)，將整個(gè)分析流程抽象成數(shù)據(jù)接入，數(shù)據(jù)處理，模型配置和報(bào)表配置4個(gè)過程。其中，模型配置完全按照OLAP模型，要求實(shí)時(shí)數(shù)據(jù)符合星型模型，存在時(shí)間戳、指標(biāo)、維度等字段。

3.2 數(shù)據(jù)處理配置

在數(shù)據(jù)處理層，實(shí)時(shí)分析平臺(tái)提供向?qū)渲庙撁?，支持用戶通過純頁面的方式就可以配置數(shù)據(jù)處理過程，這主要應(yīng)對(duì)一些簡單場(chǎng)景，針對(duì)部分連SQL都不熟悉的小白用戶提供頁面配置方案；初次之外，類似StreamingSQL，實(shí)時(shí)分析也提供用戶自定義SQL方式定義數(shù)據(jù)處理過程。

在Flink平臺(tái)化的時(shí)候，我們遇到了幾個(gè)Flink的問題，分別對(duì)其進(jìn)行了些改進(jìn)。

第一個(gè)改進(jìn)是關(guān)于checkpoint的優(yōu)雅恢復(fù)。這個(gè)問題的出發(fā)點(diǎn)是，業(yè)務(wù)希望使用Spark Streaming可以通過代碼控制從哪個(gè)checkpoint恢復(fù)，但對(duì)于Flink來講，業(yè)務(wù)沒法通過代碼控制checkpoint恢復(fù)點(diǎn)，需要手動(dòng)指定檢查點(diǎn)去恢復(fù)checkpoint。于是，我們希望Flink可以像Spark Streaming一樣，直接通過代碼方式恢復(fù)checkpoint。

針對(duì)這個(gè)問題，我們修改源碼，在Flink任務(wù)啟動(dòng)時(shí)，從實(shí)際的路徑當(dāng)中找到他最新的一個(gè)checkpoint，直接從那個(gè)checkpoint當(dāng)中恢復(fù)，當(dāng)然這個(gè)也是可以讓用戶選的，他如果還想用原生方式恢復(fù)也可以，但提供一個(gè)選項(xiàng)，它可以支持從最近的checkpoint恢復(fù)。

第二個(gè)改進(jìn)是關(guān)于Kafka Broker HA的一個(gè)問題，比如像Kafka Broker故障的時(shí)候，Kafka還可以正常工作，但Flink程序往往會(huì)掛掉。針對(duì)這個(gè)問題，我們處理了Flink在Kafka Broker退出之后的sockerTimeOutException，支持用戶重試次數(shù)配置來解決這個(gè)問題。

最后，介紹一下愛奇藝在Apache Flink的未來工作。目前StreamingSQL還只支持Spark Streaming和Structured Streaming引擎，后續(xù)很快會(huì)支持Flink引擎，大幅降低業(yè)務(wù)的Flink開發(fā)成本。隨著Flink任務(wù)規(guī)模不斷變大，我們將重點(diǎn)提升Flink在愛奇藝的成熟度，完善監(jiān)控報(bào)警，增加資源審計(jì)流程（目前還僅對(duì)Spark Streaming進(jìn)行資源審計(jì)）。另外，我們要研究下Flink 1.6的一些新特性，嘗試下Kafka 2.0，調(diào)研Exactly once方案；另外，我們將對(duì)Flink新版本進(jìn)行一些嘗試，推進(jìn)批流統(tǒng)一。