本文由 bilibili 大數(shù)據(jù)實時平臺負責(zé)人鄭志升分享，本次分享核心講解萬億級傳輸分發(fā)架構(gòu)的落地，以及 AI 領(lǐng)域如何基于 Flink 打造一套完善的預(yù)處理實時 Pipeline。

本次分享主要圍繞以下四個方面：

一、B 站實時的前世與今生

二、Flink On Yarn 的增量化管道的方案

三、Flink 和 AI 方向的一些工程實踐

四、未來的發(fā)展與思考

一、B 站實時的前世與今生

1. 生態(tài)場景輻射

說起實時計算的未來，關(guān)鍵詞就在于數(shù)據(jù)的實效性。首先從整個大數(shù)據(jù)發(fā)展的生態(tài)上，來看它的核心場景輻射：在大數(shù)據(jù)發(fā)展的初期，核心是以面向天為粒度的離線計算的場景。 那時候的數(shù)據(jù)實效性多數(shù)都是以運算以天為單位，它更加注重時間和成本的平衡。

隨著數(shù)據(jù)應(yīng)用，數(shù)據(jù)分析以及數(shù)據(jù)倉庫的普及與完善，越來越多的人對數(shù)據(jù)的實效性提出了更高的要求。比如，當(dāng)需要做一些數(shù)據(jù)的實時推薦時，數(shù)據(jù)的實效將決定它的價值。在這種情況下，整個實時計算的場景就普遍誕生。

但在實際的運作過程當(dāng)中，也遇到了很多場景 ，其實并沒有對數(shù)據(jù)有非常高的實時性要求，在這種情況下必然會存在數(shù)據(jù)從毫秒，秒或者天的新的一些場景，實時場景數(shù)據(jù)更多是以分鐘為粒度的一些增量計算的場景。對于離線計算，它更加注重成本；對實時計算，它更加注重價值實效；而對于增量計算，它更加注重去平衡成本，以及綜合的價值和時間。

2. B 站的時效性

在三個維度上，B 站的劃分是怎樣的？對于 B 站而言 ，目前有 75% 的數(shù)據(jù)是通過離線計算來進行支撐的，另外還有 20% 的場景是通過實時計算， 5% 是通過增量計算。

對于實時計算的場景， 主要是應(yīng)用在整個實時的機器學(xué)習(xí)、實時推薦、廣告搜索、數(shù)據(jù)應(yīng)用、實時渠道分析投放、報表、olap、監(jiān)控等；對于離線計算，數(shù)據(jù)輻射面廣，主要以數(shù)倉為主；對于增量計算，今年才啟動一些新的場景，比如說 binlog 的增量 Upsert 場景。

3. ETL 時效性差

對于實效性問題 ，其實早期遇到了很多痛點 ，核心集中在三個方面：

第一，傳輸管道缺乏計算能力。早期的方案，數(shù)據(jù)基本都是要按天落到 ODS ，DW 層是凌晨過后的第二天去掃描前一天所有 ODS 層的數(shù)據(jù)，也就是說，整體數(shù)據(jù)沒辦法前置清洗；第二，含有大量作業(yè)的資源集中爆發(fā)在凌晨之后，整個資源編排的壓力就會非常大；第三、實時和離線的 gap 是比較難滿足的，因為對于大部分的數(shù)據(jù)來說，純實時的成本過高，純離線的實效又太差。同時，MySQL 數(shù)據(jù)的入倉時效也不太夠。舉個例子，好比 B 站的彈幕數(shù)據(jù) ，它的體量非?？鋸?，這種業(yè)務(wù)表的同步往往需要十幾個小時，而且非常的不穩(wěn)定。

4. AI 實時工程復(fù)雜

除了實效性的問題 早期還遇到了 AI 實時工程比較復(fù)雜的問題：

第一，是整個特征工程計算效率的問題。同樣的實時特征的計算場景， 也需要在離線的場景上進行數(shù)據(jù)的回溯，計算邏輯就會重復(fù)開發(fā)；第二，整個實時鏈路比較長。一個完整的實時推薦鏈路， 涵蓋了 N 個實時和 M 個離線的十幾個作業(yè)組成，有時候遇到問題排查，整個鏈路的運維和管控成本都非常高；第三、隨著 AI 人員的增多，算法人員的投入，實驗迭代很難橫向擴展。

5. Flink 做了生態(tài)化的實踐

在這些關(guān)鍵痛點的背景下，我們集中針對 Flink 做了生態(tài)化的實踐，核心包括了整個實時數(shù)倉的應(yīng)用以及整個增量化的 ETL 管道，還有面向 AI 的機器學(xué)習(xí)的一些場景。本次的分享會更加側(cè)重增量管道以及 AI 加 Flink 的方向上。下圖展示了整體的規(guī)模，目前，整個傳輸和計算的體量，在萬億級的消息規(guī)模有 30000+ 計算核數(shù)，1000+ job 數(shù)以及 100 多個用戶。

二、Flink On Yarn 的增量化管道的方案

1. 早期的架構(gòu)

先來看一下整個管道早期的架構(gòu)，從下圖可以看出，數(shù)據(jù)其實主要是通過 Flume 來消費 Kafka 落到 HDFS。Flume 用它的事務(wù)機制，來確保數(shù)據(jù)從 Source 到 Channel， 再到 Sink 時候的一致性，最后數(shù)據(jù)落到 HDFS 之后，下游的 Scheduler 會通過掃描目錄下有沒有 tmp 文件，來判斷數(shù)據(jù)是否 Ready，以此來調(diào)度拉起下游的 ETL 離線作業(yè)。

2. 痛點

在早期遇到了不少痛點：

第一個比較關(guān)鍵的是數(shù)據(jù)質(zhì)量。最先用的是 MemoryChannel，它會存在數(shù)據(jù)的丟失，之后也試過用 FileChannel 的模式，但性能上無法達到要求。此外在 HDFS 不太穩(wěn)定的情況下，F(xiàn)lume 的事務(wù)機制就會導(dǎo)致數(shù)據(jù)會 rollback 回滾到 Channel，一定程度上會導(dǎo)致數(shù)據(jù)不斷的重復(fù)。在 HDFS 極度不穩(wěn)定的情況下，最高的重復(fù)率會達到百分位的概率；Lzo 行存儲，早期的整個傳輸是通過分隔符的形式，這種分隔符的 Schema 是比較弱約束的，而且也不支持嵌套的格式。第二點是整個數(shù)據(jù)的時效，無法提供分鐘級的查詢，因為 Flume 不像 Flink 有 Checkpoint 斬斷的機制，更多是通過 idle 機制來控制文件的關(guān)閉；第三點是下游的 ETL 聯(lián)動。前文有提到，我們更多是通過掃描 tmp 目錄是否 ready 的方案，這種情況下 scheduler 會大量的和 NameNode 調(diào)用 hadoop list 的 api，這樣會導(dǎo)致 NameNode 的壓力比較大。

3. 穩(wěn)定性相關(guān)的痛點

在穩(wěn)定性上也遇到很多問題：

第一，F(xiàn)lume 是不帶狀態(tài)的，節(jié)點異?；蛘呤侵貑⒅螅瑃mp 沒法正常關(guān)閉；第二，早期沒有依附大數(shù)據(jù)的環(huán)境，是物理部署的模式，資源伸縮很難去把控，成本也會相對偏高；第三，F(xiàn)lume 和 HDFS 在通信上有問題。比如說當(dāng)寫 HDFS 出現(xiàn)堵塞的情況，某一個節(jié)點的堵塞會反壓到 Channel，就會導(dǎo)致 Source 不會去 Kafka 消費數(shù)據(jù)，停止拉動 offset，一定程度上就會引發(fā) Kafka 的 Rebalance，最后會導(dǎo)致全局 offset 不往前推進，從而導(dǎo)致數(shù)據(jù)的堆積。

4. 萬億級的增量管道 DAG 視圖

在如上的痛點下，核心方案基于 Flink 構(gòu)建了一套萬億級的增量管道，下圖是整個運行時的 DAG 視圖。

首先，在 Flink 架構(gòu)下，KafkaSource 杜絕了 rebalance 的雪崩問題，即便整個 DAG 視圖中有某個并發(fā)度出現(xiàn)數(shù)據(jù)寫 HDFS 的堵塞，也不會導(dǎo)致全局所有 Kafka 分區(qū)的堵塞。此外的話，整個方案本質(zhì)是通過 Transform 的模塊來實現(xiàn)可擴展的節(jié)點。

第一層節(jié)點是 Parser，它主要是做數(shù)據(jù)的解壓反序列化等的解析操作；第二層是引入提供給用戶的定制化 ETL 模塊，它可以實現(xiàn)數(shù)據(jù)在管道中的定制清洗；第三層是 Exporter 模塊，它支持將數(shù)據(jù)導(dǎo)出到不同的存儲介質(zhì)。比如寫到 HDFS 時，會導(dǎo)出成 parquet；寫到 Kafka，會導(dǎo)出成 pb 格式。同時，在整個 DAG 的鏈路上引入了 ConfigBroadcast 的模塊來解決管道元數(shù)據(jù)實時更新、熱加載的問題。此外，在整個鏈路當(dāng)中，每分鐘會進行一次 checkpoint，針對增量的實際數(shù)據(jù)進行 Append，這樣就可以提供分鐘級的查詢。

5. 萬億級的增量管道整體視圖

Flink On Yarn 的整體架構(gòu)，可以看出其實整個管道視圖是劃分以 BU 為單位的。每個 Kafka 的 topic，都代表了某一種數(shù)據(jù)終端的分發(fā)，F(xiàn)link 作業(yè)就會專門負責(zé)各種終端類型的寫入處理。視圖里面還可以看到，針對 blinlog 的數(shù)據(jù)，還實現(xiàn)了整個管道的組裝，可以由多個節(jié)點來實現(xiàn)管道的運作。

6. 技術(shù)亮點

接下來來看一下整個架構(gòu)方案核心的一些技術(shù)亮點，前三個是實時功能層面的一些特色，后三個主要是在一些非功能性層面的一些優(yōu)化。

對于數(shù)據(jù)模型來說，主要是通過 parquet，利用 Protobuf 到 parquet 的映射來實現(xiàn)格式收斂；分區(qū)通知主要是因為一條管道其實是處理多條流，核心解決的是多條流數(shù)據(jù)的分區(qū) ready 的通知機制；CDC 管道更多是利用 binlog 和 HUDI 來實現(xiàn) upsert 問題的解決；小文件主要是在運行時通過 DAG 拓撲的方式來解決文件合并的問題；HDFS 通信實際是在萬億級規(guī)模下的很多種關(guān)鍵問題的優(yōu)化；最后是分區(qū)容錯的一些優(yōu)化。

6.1 數(shù)據(jù)模型

業(yè)務(wù)的開發(fā)主要是通過拼裝字符串，來組裝數(shù)據(jù)的一條條記錄的上報。后期則是通過了模型的定義和管理，以及它的開發(fā)來組織的，主要是通過在平臺的入口提供給用戶去錄制每一條流、每個表，它的 Schema ，Schema 會將它生成 Protobuf 的文件，用戶可以在平臺上去下載 Protobuf 對應(yīng)的 HDFS 模型文件，這樣，client 端的開發(fā)完全就可以通過強 Schema 方式從 pb 來進行約束。

來看一下運行時的過程，首先 Kafka 的 Source 會去消費實際上游傳過來的每一條 RawEvent 的記錄，RawEvent 里面會有 PBEvent 的對象，PBEvent 其實是一條條的 Protobuf 的記錄。數(shù)據(jù)從 Source 流到的 Parser 模塊，解析后會形成 PBEvent，PBEvent 會將用戶在平臺錄入的整個 Schema 模型，存儲在 OSS 對象系統(tǒng)上，Exporter 模塊會動態(tài)去加載模型的變更。然后通過 pb 文件去反射生成的具體事件對象，事件對象最后就可以映射落成 parquet 的格式。這里主要做了很多緩存反射的優(yōu)化，使整個 pb 的動態(tài)解析性能達到六倍的提升。最后，我們會將數(shù)據(jù)會落地到 HDFS，形成 parquet 的格式。

6.2 分區(qū)通知優(yōu)化

前面提到管道會處理上百條流，早期 Flume 的架構(gòu)，其實每個 Flume 節(jié)點，很難去感應(yīng)它自己處理的進度。同時，F(xiàn)lume 也沒辦法做到全局進度的處理。但是基于 Flink，就可以通過 Watermark 的機制來解決。

首先在 Source 會基于消息當(dāng)中的 Eventime 來生成 Watermark，Watermark 會經(jīng)過每一層的處理傳遞到 Sink，最后會通過 Commiter 模塊，以單線程的方式來匯總所有 Watermark 消息的進度。當(dāng)它發(fā)現(xiàn)全局 Watermark 已經(jīng)推進到下個小時的分區(qū)的時候，它會下發(fā)一條消息到 Hive MetStore，或者是寫入到 Kafka， 來通知上小時分區(qū)數(shù)據(jù) ready，從而可以讓下游的調(diào)度可以更快的通過消息驅(qū)動的方式來拉起作業(yè)的運行。

6.3 CDC管道上的優(yōu)化

下圖右側(cè)其實是整個 cdc 管道完整的鏈路。要實現(xiàn) MySQL 數(shù)據(jù)到 Hive 數(shù)據(jù)的完整映射，就需要解決流和批處理的問題。

首先是通過 Datax 將 MySQL 的數(shù)據(jù)全量一次性同步到的 HDFS。緊接著通過 spark 的 job，將數(shù)據(jù)初始化成 HUDI 的初始快照，接著通過 Canal 來實現(xiàn)將 Mysql 的 binlog 的數(shù)據(jù)拖到的 Kafka 的 topic，然后是通過 Flink 的 Job 將初始化快照的數(shù)據(jù)結(jié)合增量的數(shù)據(jù)進行增量更新，最后形成 HUDI 表。

整個鏈路是要解決數(shù)據(jù)的不丟不重，重點是針對 Canal 寫 Kafka 這塊，開了事務(wù)的機制，保證數(shù)據(jù)落 Kafka topic 的時候，可以做到數(shù)據(jù)在傳輸過程當(dāng)中的不丟不重。另外，數(shù)據(jù)在傳輸?shù)纳蠈悠鋵嵰灿锌赡艹霈F(xiàn)數(shù)據(jù)的重復(fù)和丟失，這時候更多是通過全局唯一 id 加毫秒級的時間戳。在整個流式 Job 中，針對全局 id 來做數(shù)據(jù)的去重，針對毫秒級時間來做數(shù)據(jù)的排序，這樣能保證數(shù)據(jù)能夠有序的更新到的 HUDI。

緊接著通過 Trace 的系統(tǒng)基于 Clickhouse 來做存儲，來統(tǒng)計各個節(jié)點數(shù)據(jù)的進出條數(shù)來做到數(shù)據(jù)的精確對比。

6.4 穩(wěn)定性 - 小文件的合并

前面提到，改造成 Flink 之后，我們是做了每分鐘的 Checkpoint，文件數(shù)的放大非常嚴(yán)重。主要是在整個 DAG 當(dāng)中去引入 merge 的 operater 來實現(xiàn)文件的合并，merge 的合并方式主要是基于并發(fā)度橫向合并，一個 writer 會對應(yīng)一個 merge。這樣每五分鐘的 Checkpoint，1 小時的 12 個文件，都會進行合并。通過種方式的話，可以將文件數(shù)極大的控制在合理的范圍內(nèi)。

6.5 HDFS 通信

實際運作過程當(dāng)中經(jīng)常會遇到整個作業(yè)堆積比較嚴(yán)重的問題，實際分析其實主是和 HDFS 通信有很大的關(guān)系。

其實 HDFS 通訊，梳理了四個關(guān)鍵的步驟：初始化 state、Invoke、Snapshot 以及 Notify Checkpoint complete。

核心問題主要發(fā)生在 Invoke 階段，Invoke 會達到文件的滾動條件，這時候會觸發(fā) flush 和 close。close 實際和 NameNode 通信的時候，會經(jīng)常出現(xiàn)堵塞的情況。

Snapshot 階段同樣會遇到一個問題，一個管道上百條流一旦觸發(fā) Snapshot，串行執(zhí)行 flush 和 close 也會非常的慢。

核心優(yōu)化集中在三個方面：

第一，減少了文件的斬斷，也就是 close 的頻次。在 Snapshot 階段，不會去 close 關(guān)閉文件，而更多的是通過文件續(xù)寫的方式。這樣，在初始化 state 的階段，就需要做文件的 Truncate 來做 Recovery 恢復(fù)。第二，是異步化 close 的改進，可以說是 close 的動作不會去堵塞整個總鏈路的處理，針對 Invoke 和 Snapshot 的 close，會將狀態(tài)管理到 state 當(dāng)中，通過初始化 state 來進行文件的恢復(fù)。第三，針對多條流，Snapshot 還做了并行化的處理，每 5 分鐘的 Checkpoint， 多條流其實就是多個 bucket，會通過循環(huán)來進行串行的處理，那么通過多線程的方式來改造，就可以減少 Checkpoint timeout 的發(fā)生。

6.6 分區(qū)容錯的一些優(yōu)化

實際在管道多條流的情況下，有些流的數(shù)據(jù)并不是每個小時都是連續(xù)的。

這種情況會帶來分區(qū)，它的 Watermark 沒有辦法正常推進，引發(fā)空分區(qū)的問題。所以我們在管道的運行過程當(dāng)中，引入 PartitionRecover 模塊，它會根據(jù) Watermark 來推進分區(qū)的通知。針對有些流的 Watermark，如果在 ideltimeout 還沒有更新的情況下，Recover 模塊來進行分區(qū)的追加。它會在每個分區(qū)的末尾到達的時候，加上 delay time 來掃描所有流的 Watermark，由此來進行兜底。

在傳輸過程當(dāng)中，當(dāng) Flink 作業(yè)重啟的時候，會遇到一波僵尸的文件，我們是通過在 DAG 的 commit 的節(jié)點，去做整個分區(qū)通知前的僵尸文件的清理刪除，來實現(xiàn)整個僵尸文件的清理，這些都屬于非功能性層面的一些優(yōu)化。

三、Flink 和 AI 方向的一些工程實踐

1. 架構(gòu)演進時間表

下圖是 AI 方向在實時架構(gòu)完整的時間線。

早在 2018 年，很多算法人員的實驗開發(fā)都是作坊式的。每個算法人員會根據(jù)自己熟悉的語言，比如說 Python，php 或 c++ 來選擇不同的語言來開發(fā)不同的實驗工程。它的維護成本非常大，而且容易出現(xiàn)故障；2019 年上半年，主要是基于 Flink 提供了 jar 包的模式來面向整個算法做一些工程的支持，可以說在整個上半年的初期，其實更多是圍繞穩(wěn)定性，通用性來做一些支持；2019 年的下半年，是通過自研的 BSQL，大大降低了模型訓(xùn)練的門檻，解決 label 以及 instance 的實時化來提高整個實驗迭代的效率；2020 年上半年，更多是圍繞整個特征的計算，流批計算打通以及特征工程效率的提升，來做一些改進；到2020 年的下半年，更多是圍繞整個實驗的流程化以及引入 AIFlow，方便的去做流批 DAG。

2. AI 工程架構(gòu)回顧

回顧一下整個 AI 工程，它的早期的架構(gòu)圖其實體現(xiàn)的是整個 AI 在 2019 年初的架構(gòu)視圖，其本質(zhì)是通過一些 single task 的方式，各種混合語言來組成的一些計算節(jié)點，來支撐著整個模型訓(xùn)練的鏈路拉起。經(jīng)過 2019 年的迭代，將整個近線的訓(xùn)練完全的替換成用 BSQL 的模式來進行開發(fā)和迭代。

3. 現(xiàn)狀痛點

在 2019 年底，其實又遇到了一些新的問題，這些問題主要集中在功能和非功能兩個維度上。

在功能層面：首先從 label 轉(zhuǎn)到產(chǎn)生 instance 流，以及到模型訓(xùn)練，到線上預(yù)測，乃至真正的實驗效果，整個鏈路非常的長且復(fù)雜；第二，整個實時的特征、離線特征、以及流批的一體，涉及到非常多的作業(yè)組成，整個鏈路很復(fù)雜。同時實驗和 online 都要做特征的計算，結(jié)果不一致會導(dǎo)致最終的效果出現(xiàn)問題。此外，特征存在哪里也不好找，沒辦法去追溯。

在非功能性層面，算法的同學(xué)經(jīng)常會遇到，不知道 Checkpoint 是什么，要不要開，有啥配置。此外，線上出問題的時候也不好排查，整個鏈路都非常的長。所以第三點就是，完整的實驗進度需要涉及的資源是非常多的，但是對算法來說它根本就不知道這些資源是什么以及需要多少，這些問題其實都都對算法產(chǎn)生很大的困惑。

4. 痛點歸結(jié)

歸根結(jié)底，集中在三個方面：

第一是一致性的問題。從數(shù)據(jù)的預(yù)處理，到模型訓(xùn)練，再到預(yù)測，各個環(huán)節(jié)其實是斷層的。當(dāng)中包括數(shù)據(jù)的不一致，也包括計算邏輯的不一致；第二，整個實驗迭代非常慢。一個完整的實驗鏈路，其實對算法同學(xué)來說，他需要掌握東西非常多。同時實驗背后的物料沒辦法進行共享。比如說有些特征，每個實驗背后都要重復(fù)開發(fā)；第三，是運維和管控的成本比較高。

完整的實驗鏈路，背后其實是包含實時的一條工程加離線的一條工程鏈路組成，線上的問題很難去排查。

5. 實時 AI 工程的雛形

在這樣的一些痛點下，在 20 年主要是集中在 AI 方向上去打造實時工程的雛形。核心是通過下面三個方面來進行突破。

第一是在 BSQL 的一些能力上，對于算法，希望通過面向 SQL 來開發(fā)以此降低工程投入；第二是特征工程，會通過核心解決特征計算的一些問題來滿足特征的一些支持；第三是整個實驗的協(xié)作，算法的目的其實在于實驗，希望去打造一套端到端的實驗協(xié)作，最終希望做到面向算法能夠“一鍵實驗”。

6. 特征工程-難點

我們在特征工程中遇到了一些難點。

第一是在實時特征計算上，因為它需要將結(jié)果利用到整個線上的預(yù)測服務(wù)，所以它對延遲以及穩(wěn)定性的要求都非常的高；第二是整個實時和離線的計算邏輯一致，我們經(jīng)常遇到一個實時特征，它需要去回溯過去 30 天到到 60 天的離線數(shù)據(jù)，怎么做到實時特征的計算邏輯能同樣在離線特征的計算上去復(fù)用；第三是整個離線特征的流批一體比較難打通。實時特征的計算邏輯經(jīng)常會帶有窗口時序等等一些流式的概念，但是離線特征是沒有這些語義的。

7. 實時特征

這里看一下我們怎么去做實時特征，圖中的右側(cè)是最典型的一些場景。比如說我要實時統(tǒng)計用戶最近一分鐘、6 小時、12 小時、24 小時，對各個 UP 主相關(guān)視頻的播放次數(shù)。針對這樣場景，其實里面有兩個點：

第一、它需要用到滑動窗口來做整個用戶過去歷史的計算。此外，數(shù)據(jù)在滑動計算過程當(dāng)中，它還需要去關(guān)聯(lián) UP 主的一些基礎(chǔ)的信息維表，來獲取 UP 主的一些視頻來統(tǒng)計他的播放次數(shù)。歸根結(jié)底，其實遇到了兩個比較大的痛。用 Flink 原生的滑動窗口，分鐘級的滑動，會導(dǎo)致窗口比較多，性能會損耗比較大。同時細粒度的窗口也會導(dǎo)致定時器過多，清理效率比較差。第二是維表查詢，會遇到是多個 key 要去查詢 HBASE 的多個對應(yīng)的 value，這種情況需要去支持?jǐn)?shù)組的并發(fā)查詢。

在兩個痛點下，針對滑動窗口，主要是改造成為 Group By 的模式，加上 agg 的 UDF 的模式，將整個一小時、六小時、十二小時、二十四小時的一些窗口數(shù)據(jù)，存放到整個 Rocksdb 當(dāng)中。這樣通過 UDF 模式，整個數(shù)據(jù)觸發(fā)機制就可以基于 Group By 實現(xiàn)記錄級的觸發(fā)，整個語義、時效性都會提升的比較大。同時在整個 AGG 的 UDF 函數(shù)當(dāng)中，通過 Rocksdb 來做 state，在 UDF 當(dāng)中來維護數(shù)據(jù)的生命周期。此外還擴展了整個 SQL 實現(xiàn)了數(shù)組級別的維表查詢。最后的整個效果其實可以在實時特征的方向上，通過超大窗口的模式來支持各種計算場景。

8. 特征-離線

接下來看一下離線，左側(cè)視圖上半部分是完整的實時特征的計算鏈路，可以看出要解決同樣的一條 SQL，在離線的計算上也能夠復(fù)用，那就需要去解決相應(yīng)的一些計算的 IO 都能夠復(fù)用的問題。比如在流式上是通過 Kafka 來進行數(shù)據(jù)的輸入，在離線上需要通過 HDFS 來做數(shù)據(jù)的輸入。在流式上是通過 KFC 或者 AVBase 等等的一些 kv 引擎來支持，在離線上就需要通過 hive 引擎來解決，歸根結(jié)底，其實需要去解決三個方面的問題：

第一，需要去模擬整個流式消費的能力，能夠支持在離線的場景下去消費 HDFS 數(shù)據(jù)；第二，需要解決 HDFS 數(shù)據(jù)在消費過程當(dāng)中的分區(qū)有序的問題，類似 Kafka 的分區(qū)消費；第三，需要去模擬 kv 引擎維表化的消費，實現(xiàn)基于 hive 的維表消費。還需要解決一個問題，當(dāng)從 HDFS 拉取的每一條記錄，每一條記錄其實消費 hive 表的時候都有對應(yīng)的 Snapshot，就相當(dāng)于是每一條數(shù)據(jù)的時間戳，要消費對應(yīng)數(shù)據(jù)時間戳的分區(qū)。

9. 優(yōu)化

9.1 離線-分區(qū)有序

分區(qū)有序的方案其實主要是基于數(shù)據(jù)在落 HDFS 時候，前置做了一些改造。首先數(shù)據(jù)在落 HDFS 之前，是傳輸?shù)墓艿溃ㄟ^ Kafka 消費數(shù)據(jù)。在 Flink 的作業(yè)從 Kafka 拉取數(shù)據(jù)之后，通過 Eventtime 去提取數(shù)據(jù)的 watermark，每一個 Kafka Source 的并發(fā)度會將 watermark 匯報到 JobManager 當(dāng)中的 GlobalWatermark 模塊，GlobalAgg 會匯總來自每一個并發(fā)度 Watermark 推進的進度，從而去統(tǒng)計 GlobalWatermark 的進展。根據(jù) GlobalWatermark 的進展來計算出當(dāng)中有哪些并發(fā)度的 Watermark 計算過快的問題，從而通過 GlobalAgg 下發(fā)給 Kafka Source 控制信息，Kafka Source 有些并發(fā)度過快的情況下，它的整個分區(qū)推進就降低速度。這樣，在 HDFS Sink 模塊，在同時間片上收到的數(shù)據(jù)記錄的整個 Event time 基本上有序的，最終落到 HDFS 還會在文件名上去標(biāo)識它相應(yīng)的分區(qū)以及相應(yīng)的時間片范圍。最后在 HDFS 分區(qū)目錄下，就可以實現(xiàn)數(shù)據(jù)分區(qū)的有序目錄。

9.2 離線-分區(qū)增量消費

數(shù)據(jù)在 HDFS 增量有序之后，實現(xiàn)了 HDFStreamingSource，它會針對文件做 Fecher 分區(qū)，針對每個文件都有 Fecher 的線程，且每個 Fecher 線程會統(tǒng)計每一個文件。它 offset 處理了游標(biāo)的進度，會將狀態(tài)根據(jù) Checkpoint 的過程，將它更新到的 State 當(dāng)中。

這樣就可以實現(xiàn)整個文件消費的有序推進。在回溯歷史數(shù)據(jù)的時候，離線作業(yè)就會涉及到整個作業(yè)的停止。實際是在整個 FileFetcher 的模塊當(dāng)中去引入一個分區(qū)結(jié)束的標(biāo)識，且會在每一個線程去統(tǒng)計每一個分區(qū)的時候，去感應(yīng)它分區(qū)的結(jié)束，分區(qū)結(jié)束后的狀態(tài)最后匯總到的 cancellationManager，并進一步會匯總到 Job Manager 去更新全局分區(qū)的進度，當(dāng)全局所有的分區(qū)都到了末尾的游標(biāo)時候，會將整個 Flink 作業(yè)進行 cancel 關(guān)閉掉。

9.3 離線 - Snapshot 維表

前面講到整個離線數(shù)據(jù)，其實數(shù)據(jù)都在 hive 上，hive 的 HDFS 表數(shù)據(jù)的整個表字段信息會非常的多，但實際做離線特征的時候，需要的信息其實是很少的，因此需要在 hive 的過程先做離線字段裁剪，將一張 ODS 的表清洗成 DW 的表，DW 的表會最后通過 Flink 運行 Job，內(nèi)部會有個 reload 的 scheduler，它會定期的去根據(jù)數(shù)據(jù)當(dāng)前推進的 Watermark 的分區(qū)，去拉取在 hive 當(dāng)中每一個分區(qū)對應(yīng)的表信息。通過去下載某 HDFS 的 hive 目錄當(dāng)中的一些數(shù)據(jù)，最后會在整個內(nèi)存當(dāng)中 reload 成 Rocksdb 的文件，Rocksdb 其實就是最后用來提供維表 KV 查詢的組件。

組件里面會包含多個 Rocksdb 的 build 構(gòu)建過程，主要是取決于整個數(shù)據(jù)流動的過程當(dāng)中的 Eventtime，如果發(fā)現(xiàn) Eventtime 推進已經(jīng)快到小時分區(qū)結(jié)束的末尾時候，會通過懶加載的模式去主動 reload，構(gòu)建下一個小時 Rocksdb 的分區(qū)，通過這種方式，來切換整個 Rocksdb 的讀取。

10. 實驗流批一體

在上面三個優(yōu)化，也就是分區(qū)有序增量，類 Kafka 分區(qū) Fetch 消費，以及維表 Snapshot 的基礎(chǔ)下，最終是實現(xiàn)了實時特征和離線特征，共用一套 SQL 的方案，打通了特征的流批計算。緊接著來看一下整個實驗，完整的流批一體的鏈路，從圖中可以看出最上面的粒度是整個離線的完整的計算過程。第二是整個近線的過程，離線過程其實所用計算的語義都是和近線過程用實時消費的語義是完全一致的，都是用 Flink 來提供 SQL 計算的。

來看一下近線，其實 Label join 用的是 Kafka 的一條點擊流以及展現(xiàn)流，到了整個離線的計算鏈路，則用的一條 HDFS 點擊的目錄和 HDFS 展現(xiàn)目錄。特征數(shù)據(jù)處理也是一樣的，實時用的是 Kafka 的播放數(shù)據(jù)，以及 Hbase 的一些稿件數(shù)據(jù)。對于離線來說，用的是 hive 的稿件數(shù)據(jù)，以及 hive 的播放數(shù)據(jù)。除了整個離線和近線的流批打通，還將整個近線產(chǎn)生的實時的數(shù)據(jù)效果匯總到 OLAP 引擎上，通過 superset 來提供整個實時的指標(biāo)可視化。其實從圖可以看出完整的復(fù)雜流批一體的計算鏈路，當(dāng)中包含的計算節(jié)點是非常的復(fù)雜和龐多的。

11. 實驗協(xié)作 - 挑戰(zhàn)

下階段挑戰(zhàn)更多是在實驗協(xié)作上，下圖是將前面整個鏈路進行簡化后的抽象。從圖中可以看出，三個虛線的區(qū)域框內(nèi)，分別是離線的鏈路加兩個實時的鏈路，三個完整的鏈路構(gòu)成作業(yè)的流批，實際上就是一個工作流最基本的過程。里面需要去完成工作流完整的抽象，包括了流批事件的驅(qū)動機制，以及，對于算法在 AI 領(lǐng)域上更多希望用 Python 來定義完整的 flow，此外還將整個輸入，輸出以及它的整個計算趨于模板化，這樣可以做到方便整個實驗的克隆。

12. 引入 AIFlow

整個工作流上在下半年更多是和社區(qū)合作，引入了 AIFlow 的整套方案。

右側(cè)其實是整個 AIFlow 完整鏈路的DAG視圖，可以看出整個節(jié)點，其實它支持的類型是沒有任何限制的，可以是流式節(jié)點，也可以是離線節(jié)點。此外的話，整個節(jié)點與節(jié)點之間通信的邊是可以支持?jǐn)?shù)據(jù)驅(qū)動以及事件驅(qū)動的。引入 AIFlow 的好處主要在于，AIFlow 提供基于 Python 語義來方便去定義完整的 AIFlow 的工作流，同時還包括整個工作流的進度的調(diào)度。

在節(jié)點的邊上，相比原生的業(yè)界的一些 Flow 方案，他還支持基于事件驅(qū)動的整個機制。好處是可以幫助在兩個 Flink 作業(yè)之間，通過 Flink 當(dāng)中 watermark 處理數(shù)據(jù)分區(qū)的進度去下發(fā)一條事件驅(qū)動的消息來拉起下一個離線或者實時的作業(yè)。

此外還支持周邊的一些配套服務(wù)，包括通知的一些消息模塊服務(wù)，還有元數(shù)據(jù)的服務(wù)，以及在 AI 領(lǐng)域一些模型中心的服務(wù)。

13. Python 定義 Flow

來看一下基于 AIFlow 是如何最終定義成 Python 的工作流。右邊的視圖是一個線上項目的完整工作流的定義。第一、是整個是 Spark job 的定義，當(dāng)中通過配置 dependence 來描述整個下游的依賴關(guān)系，它會下發(fā)一條事件驅(qū)動的消息來拉起下面的 Flink 流式作業(yè)。流式作業(yè)也同樣可以通過消息驅(qū)動的方式來拉起下面的 Spark 作業(yè)。整個語義的定義非常的簡單，只需要四個步驟，配置每節(jié)點的 confg 的信息，以及定義每節(jié)點的 operation 的行為，還有它的 dependency 的依賴，最后去運行整個 flow 的拓撲視圖。

14. 基于事件驅(qū)動流批

接下來看一下完整的流批調(diào)度的驅(qū)動機制，下圖右側(cè)是完整的三個工作節(jié)點的驅(qū)動視圖。第一個是從 Source 到 SQL 到 Sink。引入的黃色方框是擴展的 supervisor，他可以收集全局的 watermark 進度。當(dāng)整個流式作業(yè)發(fā)現(xiàn) watermark 可以推進到下一個小時的分區(qū)的時候，它會下發(fā)一條消息，去給到 NotifyService。NotifyService 拿到這條消息之后，它會去下發(fā)給到下一個作業(yè)，下一個作業(yè)主要會在整個 Flink 的 DAG 當(dāng)中去引入 flow 的 operator，operator 在沒有收到上個作業(yè)下發(fā)了消息之前，它會堵塞整個作業(yè)的運行。直到收到消息驅(qū)動之后，就代表上游其實上一個小時分區(qū)已經(jīng)完成了，這時下個 flow 節(jié)點就可以驅(qū)動拉起來運作。同樣，下個工作流節(jié)點也引入了 GlobalWatermark Collector 的模塊來匯總收集它的處理的進度。當(dāng)上一個小時分區(qū)完成之后，它也會下發(fā)一條消息到 NotifyService，NotifyService 會將這條消息去驅(qū)動調(diào)用 AIScheduler 的模塊，從而去拉起 spark 離線作業(yè)來做 spark 離線的收尾。從里你們可以看出，整個鏈路其實是支持批到批，批到流以及流到流，以及流到批的四個場景。

15. 實時 AI 全鏈路的雛形

在流和批的整個 flow 定義和調(diào)度的基礎(chǔ)上，在 2020 年初步構(gòu)建出來了實時 AI 全鏈路的雛形，核心是面向?qū)嶒?。算法同學(xué)也可以基于 SQL 來開發(fā)的 Node 的節(jié)點，Python 是可以定義完整的 DAG 工作流。監(jiān)控，告警以及運維是一體化的。

同時，支持從離線到實時的打通，從數(shù)據(jù)處理到模型訓(xùn)練，從模型訓(xùn)練到實驗效果的打通，以及面向端到端的打通。右側(cè)是整個近線實驗的鏈路。下面是將整個實驗鏈路產(chǎn)出的物料數(shù)據(jù)提供給在線的預(yù)測訓(xùn)練的服務(wù)。整體會有三個方面的配套：

一是基礎(chǔ)的一些平臺功能，包括實驗管理，模型管理，特征管理等等；其次也包括整個 AIFlow 底層的一些 service 的服務(wù)；再有是一些平臺級的 metadata 的元數(shù)據(jù)服務(wù)。

四、未來的一些展望

在未來的一年，我們還會更加集中在兩個方面的一些工作。

第一是數(shù)據(jù)湖的方向上，會集中在 ODS 到 DW 層的一些增量計算場景，以及 DW 到 ADS 層的一些場景的突破，核心會結(jié)合 Flink 加 Iceberg 以及 HUDI 來作為該方向的落地。在實時 AI 平臺上，會進一步去面向?qū)嶒瀬硖峁┮惶讓崟r的 AI 協(xié)作平臺，核心是希望打造高效，能夠提煉簡化算法人員的工程平臺。