本文由 bilibili 大資料實時平臺負責人鄭志升分享，本次分享核心講解萬億級傳輸分發架構的落地，以及 AI 領域如何基於 Flink 打造一套完善的預處理實時 Pipeline。本次分享主要圍繞以下四個方面：

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

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

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

四、未來的發展與思考

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

1. 生態場景輻射

說起實時計算的未來，關鍵詞就在於資料的實效性。首先從整個大資料發展的生態上，來看它的核心場景輻射：在大資料發展的初期，核心是以面向天為粒度的離線計算的場景。 那時候的資料實效性多數都是以運算以天為單位，它更加註重時間和成本的平衡。

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

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

2. B 站的時效性

在三個維度上，B 站的劃分是怎樣的？對於 B 站而言 ，目前有 75% 的資料是透過離線計算來進行支撐的，另外還有 20% 的場景是透過實時計算， 5% 是透過增量計算。

對於實時計算的場景， 主要是應用在整個實時的機器學習、實時推薦、廣告搜尋、資料應用、實時渠道分析投放、報表、olap、監控等；

對於離線計算，資料輻射面廣，主要以數倉為主；

對於增量計算，今年才啟動一些新的場景，比如說 binlog 的增量 Upsert 場景。

3. ETL 時效性差

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

第一，傳輸管道缺乏計算能力。早期的方案，資料基本都是要按天落到 ODS ，DW 層是凌晨過後的第二天去掃描前一天所有 ODS 層的資料，也就是說，整體資料沒辦法前置清洗；

第二，含有大量作業的資源集中爆發在凌晨之後，整個資源編排的壓力就會非常大；

第三、實時和離線的 gap 是比較難滿足的，因為對於大部分的資料來說，純實時的成本過高，純離線的實效又太差。同時，MySQL 資料的入倉時效也不太夠。舉個例子，好比 B 站的彈幕資料 ，它的體量非常誇張，這種業務表的同步往往需要十幾個小時，而且非常的不穩定。

4. AI 實時工程複雜

除了實效性的問題 早期還遇到了 AI 實時工程比較複雜的問題：

第一，是整個特徵工程計算效率的問題。同樣的實時特徵的計算場景， 也需要在離線的場景上進行資料的回溯，計算邏輯就會重複開發；

第二，整個實時鏈路比較長。一個完整的實時推薦鏈路， 涵蓋了 N 個實時和 M 個離線的十幾個作業組成，有時候遇到問題排查，整個鏈路的運維和管控成本都非常高；

第三、隨著 AI 人員的增多，演算法人員的投入，實驗迭代很難橫向擴充套件。

5. Flink 做了生態化的實踐

在這些關鍵痛點的背景下，我們集中針對 Flink 做了生態化的實踐，核心包括了整個實時數倉的應用以及整個增量化的 ETL 管道，還有面向 AI 的機器學習的一些場景。本次的分享會更加側重增量管道以及 AI 加 Flink 的方向上。下圖展示了整體的規模，目前，整個傳輸和計算的體量，在萬億級的訊息規模有 30000+ 計算核數，1000+ job 數以及 100 多個使用者。

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

1. 早期的架構

先來看一下整個管道早期的架構，從下圖可以看出，資料其實主要是透過 Flume 來消費 Kafka 落到 HDFS。Flume 用它的事務機制，來確保資料從 Source 到 Channel， 再到 Sink 時候的一致性，最後資料落到 HDFS 之後，下游的 Scheduler 會透過掃描目錄下有沒有 tmp 檔案，來判斷資料是否 Ready，以此來排程拉起下游的 ETL 離線作業。

2. 痛點

在早期遇到了不少痛點：

第一個比較關鍵的是資料質量。最先用的是 MemoryChannel，它會存在資料的丟失，之後也試過用 FileChannel 的模式，但效能上無法達到要求。此外在 HDFS 不太穩定的情況下，Flume 的事務機制就會導致資料會 rollback 回滾到 Channel，一定程度上會導致資料不斷的重複。在 HDFS 極度不穩定的情況下，最高的重複率會達到百分位的機率；Lzo 行儲存，早期的整個傳輸是透過分隔符的形式，這種分隔符的 Schema 是比較弱約束的，而且也不支援巢狀的格式。

第二點是整個資料的時效，無法提供分鐘級的查詢，因為 Flume 不像 Flink 有 Checkpoint 斬斷的機制，更多是透過 idle 機制來控制檔案的關閉；

第三點是下游的 ETL 聯動。前文有提到，我們更多是透過掃描 tmp 目錄是否 ready 的方案，這種情況下 scheduler 會大量的和 NameNode 呼叫 hadoop list 的 api，這樣會導致 NameNode 的壓力比較大。

3. 穩定性相關的痛點

在穩定性上也遇到很多問題：

第一，Flume 是不帶狀態的，節點異常或者是重啟之後，tmp 沒法正常關閉；

第二，早期沒有依附大資料的環境，是物理部署的模式，資源伸縮很難去把控，成本也會相對偏高；

第三，Flume 和 HDFS 在通訊上有問題。比如說當寫 HDFS 出現堵塞的情況，某一個節點的堵塞會反壓到 Channel，就會導致 Source 不會去 Kafka 消費資料，停止拉動 offset，一定程度上就會引發 Kafka 的 Rebalance，最後會導致全域性 offset 不往前推進，從而導致資料的堆積。

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

在如上的痛點下，核心方案基於 Flink 構建了一套萬億級的增量管道，下圖是整個執行時的 DAG 檢視。

首先，在 Flink 架構下，KafkaSource 杜絕了 rebalance 的雪崩問題，即便整個 DAG 檢視中有某個併發度出現數據寫 HDFS 的堵塞，也不會導致全域性所有 Kafka 分割槽的堵塞。此外的話，整個方案本質是透過 Transform 的模組來實現可擴充套件的節點。

第一層節點是 Parser，它主要是做資料的解壓反序列化等的解析操作；

第二層是引入提供給使用者的定製化 ETL 模組，它可以實現資料在管道中的定製清洗；

第三層是 Exporter 模組，它支援將資料匯出到不同的儲存介質。比如寫到 HDFS 時，會匯出成 parquet；寫到 Kafka，會匯出成 pb 格式。同時，在整個 DAG 的鏈路上引入了 ConfigBroadcast 的模組來解決管道元資料實時更新、熱載入的問題。此外，在整個鏈路當中，每分鐘會進行一次 checkpoint，針對增量的實際資料進行 Append，這樣就可以提供分鐘級的查詢。

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

Flink On Yarn 的整體架構，可以看出其實整個管道檢視是劃分以 BU 為單位的。每個 Kafka 的 topic，都代表了某一種資料終端的分發，Flink 作業就會專門負責各種終端型別的寫入處理。視圖裡面還可以看到，針對 blinlog 的資料，還實現了整個管道的組裝，可以由多個節點來實現管道的運作。

6. 技術亮點

接下來來看一下整個架構方案核心的一些技術亮點，前三個是實時功能層面的一些特色，後三個主要是在一些非功能性層面的一些最佳化。

對於資料模型來說，主要是透過 parquet，利用 Protobuf 到 parquet 的對映來實現格式收斂；

分割槽通知主要是因為一條管道其實是處理多條流，核心解決的是多條流資料的分割槽 ready 的通知機制；

CDC 管道更多是利用 binlog 和 HUDI 來實現 upsert 問題的解決；

小檔案主要是在執行時透過 DAG 拓撲的方式來解決檔案合併的問題；

HDFS 通訊實際是在萬億級規模下的很多種關鍵問題的最佳化；

最後是分割槽容錯的一些最佳化。

6。1 資料模型

業務的開發主要是透過拼裝字串，來組裝資料的一條條記錄的上報。後期則是通過了模型的定義和管理，以及它的開發來組織的，主要是透過在平臺的入口提供給使用者去錄製每一條流、每個表，它的 Schema ，Schema 會將它生成 Protobuf 的檔案，使用者可以在平臺上去下載 Protobuf 對應的 HDFS 模型檔案，這樣，client 端的開發完全就可以透過強 Schema 方式從 pb 來進行約束。

來看一下執行時的過程，首先 Kafka 的 Source 會去消費實際上游傳過來的每一條 RawEvent 的記錄，RawEvent 裡面會有 PBEvent 的物件，PBEvent 其實是一條條的 Protobuf 的記錄。資料從 Source 流到的 Parser 模組，解析後會形成 PBEvent，PBEvent 會將使用者在平臺錄入的整個 Schema 模型，儲存在 OSS 物件系統上，Exporter 模組會動態去載入模型的變更。然後透過 pb 檔案去反射生成的具體事件物件，事件物件最後就可以對映落成 parquet 的格式。這裡主要做了很多快取反射的最佳化，使整個 pb 的動態解析效能達到六倍的提升。最後，我們會將資料會落地到 HDFS，形成 parquet 的格式。

6。2 分割槽通知最佳化

前面提到管道會處理上百條流，早期 Flume 的架構，其實每個 Flume 節點，很難去感應它自己處理的進度。同時，Flume 也沒辦法做到全域性進度的處理。但是基於 Flink，就可以透過 Watermark 的機制來解決。

首先在 Source 會基於訊息當中的 Eventime 來生成 Watermark，Watermark 會經過每一層的處理傳遞到 Sink，最後會透過 Commiter 模組，以單執行緒的方式來彙總所有 Watermark 訊息的進度。當它發現全域性 Watermark 已經推進到下個小時的分割槽的時候，它會下發一條訊息到 Hive MetStore，或者是寫入到 Kafka， 來通知上小時分割槽資料 ready，從而可以讓下游的排程可以更快的透過訊息驅動的方式來拉起作業的執行。

6。3 CDC管道上的最佳化

下圖右側其實是整個 cdc 管道完整的鏈路。要實現 MySQL 資料到 Hive 資料的完整對映，就需要解決流和批處理的問題。

首先是透過 Datax 將 MySQL 的資料全量一次性同步到的 HDFS。緊接著透過 spark 的 job，將資料初始化成 HUDI 的初始快照，接著透過 Canal 來實現將 Mysql 的 binlog 的資料拖到的 Kafka 的 topic，然後是透過 Flink 的 Job 將初始化快照的資料結合增量的資料進行增量更新，最後形成 HUDI 表。

整個鏈路是要解決資料的不丟不重，重點是針對 Canal 寫 Kafka 這塊，開了事務的機制，保證資料落 Kafka topic 的時候，可以做到資料在傳輸過程當中的不丟不重。另外，資料在傳輸的上層其實也有可能出現數據的重複和丟失，這時候更多是透過全域性唯一 id 加毫秒級的時間戳。在整個流式 Job 中，針對全域性 id 來做資料的去重，針對毫秒級時間來做資料的排序，這樣能保證資料能夠有序的更新到的 HUDI。

緊接著透過 Trace 的系統基於 Clickhouse 來做儲存，來統計各個節點資料的進出條數來做到資料的精確對比。

6。4 穩定性 - 小檔案的合併

前面提到，改造成 Flink 之後，我們是做了每分鐘的 Checkpoint，檔案數的放大非常嚴重。主要是在整個 DAG 當中去引入 merge 的 operater 來實現檔案的合併，merge 的合併方式主要是基於併發度橫向合併，一個 writer 會對應一個 merge。這樣每五分鐘的 Checkpoint，1 小時的 12 個檔案，都會進行合併。透過種方式的話，可以將檔案數極大的控制在合理的範圍內。

6。5 HDFS 通訊

實際運作過程當中經常會遇到整個作業堆積比較嚴重的問題，實際分析其實主是和 HDFS 通訊有很大的關係。

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

核心問題主要發生在 Invoke 階段，Invoke 會達到檔案的捲軸件，這時候會觸發 flush 和 close。close 實際和 NameNode 通訊的時候，會經常出現堵塞的情況。

Snapshot 階段同樣會遇到一個問題，一個管道上百條流一旦觸發 Snapshot，序列執行 flush 和 close 也會非常的慢。

核心最佳化集中在三個方面：

第一，減少了檔案的斬斷，也就是 close 的頻次。在 Snapshot 階段，不會去 close 關閉檔案，而更多的是透過檔案續寫的方式。這樣，在初始化 state 的階段，就需要做檔案的 Truncate 來做 Recovery 恢復。

第二，是非同步化 close 的改進，可以說是 close 的動作不會去堵塞整個總鏈路的處理，針對 Invoke 和 Snapshot 的 close，會將狀態管理到 state 當中，透過初始化 state 來進行檔案的恢復。

第三，針對多條流，Snapshot 還做了並行化的處理，每 5 分鐘的 Checkpoint， 多條流其實就是多個 bucket，會透過迴圈來進行序列的處理，那麼透過多執行緒的方式來改造，就可以減少 Checkpoint timeout 的發生。

6。6 分割槽容錯的一些最佳化

實際在管道多條流的情況下，有些流的資料並不是每個小時都是連續的。

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

在傳輸過程當中，當 Flink 作業重啟的時候，會遇到一波殭屍的檔案，我們是透過在 DAG 的 commit 的節點，去做整個分割槽通知前的殭屍檔案的清理刪除，來實現整個殭屍檔案的清理，這些都屬於非功能性層面的一些最佳化。

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

1. 架構演進時間表

下圖是 AI 方向在實時架構完整的時間線。

早在 2018 年，很多演算法人員的實驗開發都是作坊式的。每個演算法人員會根據自己熟悉的語言，比如說 Python，php 或 c++ 來選擇不同的語言來開發不同的實驗工程。它的維護成本非常大，而且容易出現故障；

2019 年上半年，主要是基於 Flink 提供了 jar 包的模式來面向整個演算法做一些工程的支援，可以說在整個上半年的初期，其實更多是圍繞穩定性，通用性來做一些支援；

2019 年的下半年，是透過自研的 BSQL，大大降低了模型訓練的門檻，解決 label 以及 instance 的實時化來提高整個實驗迭代的效率；

2020 年上半年，更多是圍繞整個特徵的計算，流批計算打通以及特徵工程效率的提升，來做一些改進；

到2020 年的下半年，更多是圍繞整個實驗的流程化以及引入 AIFlow，方便的去做流批 DAG。

2. AI 工程架構回顧

回顧一下整個 AI 工程，它的早期的架構圖其實體現的是整個 AI 在 2019 年初的架構檢視，其本質是透過一些 single task 的方式，各種混合語言來組成的一些計算節點，來支撐著整個模型訓練的鏈路拉起。經過 2019 年的迭代，將整個近線的訓練完全的替換成用 BSQL 的模式來進行開發和迭代。

3. 現狀痛點

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

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

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

4. 痛點歸結

歸根結底，集中在三個方面：

第一是一致性的問題。從資料的預處理，到模型訓練，再到預測，各個環節其實是斷層的。當中包括資料的不一致，也包括計算邏輯的不一致；

第二，整個實驗迭代非常慢。一個完整的實驗鏈路，其實對演算法同學來說，他需要掌握東西非常多。同時實驗背後的物料沒辦法進行共享。比如說有些特徵，每個實驗背後都要重複開發；

第三，是運維和管控的成本比較高。

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

5. 實時 AI 工程的雛形

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

第一是在 BSQL 的一些能力上，對於演算法，希望透過面向 SQL 來開發以此降低工程投入；

第二是特徵工程，會透過核心解決特徵計算的一些問題來滿足特徵的一些支援；

第三是整個實驗的協作，演算法的目的其實在於實驗，希望去打造一套端到端的實驗協作，最終希望做到面向演算法能夠“一鍵實驗”。

6. 特徵工程-難點

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

第一是在實時特徵計算上，因為它需要將結果利用到整個線上的預測服務，所以它對延遲以及穩定性的要求都非常的高；

第二是整個實時和離線的計算邏輯一致，我們經常遇到一個實時特徵，它需要去回溯過去 30 天到到 60 天的離線資料，怎麼做到實時特徵的計算邏輯能同樣在離線特徵的計算上去複用；

第三是整個離線特徵的流批一體比較難打通。實時特徵的計算邏輯經常會帶有視窗時序等等一些流式的概念，但是離線特徵是沒有這些語義的。

7. 實時特徵

這裡看一下我們怎麼去做實時特徵，圖中的右側是最典型的一些場景。比如說我要實時統計使用者最近一分鐘、6 小時、12 小時、24 小時，對各個 UP 主相關影片的播放次數。針對這樣場景，其實裡面有兩個點：

第一、它需要用到滑動視窗來做整個使用者過去歷史的計算。此外，資料在滑動計算過程當中，它還需要去關聯 UP 主的一些基礎的資訊維表，來獲取 UP 主的一些影片來統計他的播放次數。歸根結底，其實遇到了兩個比較大的痛。用 Flink 原生的滑動視窗，分鐘級的滑動，會導致視窗比較多，效能會損耗比較大。同時細粒度的視窗也會導致定時器過多，清理效率比較差。

第二是維表查詢，會遇到是多個 key 要去查詢 HBASE 的多個對應的 value，這種情況需要去支援陣列的併發查詢。

在兩個痛點下，針對滑動視窗，主要是改造成為 Group By 的模式，加上 agg 的 UDF 的模式，將整個一小時、六小時、十二小時、二十四小時的一些視窗資料，存放到整個 Rocksdb 當中。這樣透過 UDF 模式，整個資料觸發機制就可以基於 Group By 實現記錄級的觸發，整個語義、時效性都會提升的比較大。同時在整個 AGG 的 UDF 函式當中，透過 Rocksdb 來做 state，在 UDF 當中來維護資料的生命週期。此外還擴充套件了整個 SQL 實現了陣列級別的維表查詢。最後的整個效果其實可以在實時特徵的方向上，透過超大視窗的模式來支援各種計算場景。

8. 特徵-離線

接下來看一下離線，左側檢視上半部分是完整的實時特徵的計算鏈路，可以看出要解決同樣的一條 SQL，在離線的計算上也能夠複用，那就需要去解決相應的一些計算的 IO 都能夠複用的問題。比如在流式上是透過 Kafka 來進行資料的輸入，在離線上需要透過 HDFS 來做資料的輸入。在流式上是透過 KFC 或者 AVBase 等等的一些 kv 引擎來支援，在離線上就需要透過 hive 引擎來解決，歸根結底，其實需要去解決三個方面的問題：

第一，需要去模擬整個流式消費的能力，能夠支援在離線的場景下去消費 HDFS 資料；

第二，需要解決 HDFS 資料在消費過程當中的分割槽有序的問題，類似 Kafka 的分割槽消費；

第三，需要去模擬 kv 引擎維表化的消費，實現基於 hive 的維表消費。還需要解決一個問題，當從 HDFS 拉取的每一條記錄，每一條記錄其實消費 hive 表的時候都有對應的 Snapshot，就相當於是每一條資料的時間戳，要消費對應資料時間戳的分割槽。

9. 最佳化

9。1 離線-分割槽有序

分割槽有序的方案其實主要是基於資料在落 HDFS 時候，前置做了一些改造。首先資料在落 HDFS 之前，是傳輸的管道，透過 Kafka 消費資料。在 Flink 的作業從 Kafka 拉取資料之後，透過 Eventtime 去提取資料的 watermark，每一個 Kafka Source 的併發度會將 watermark 彙報到 JobManager 當中的 GlobalWatermark 模組，GlobalAgg 會彙總來自每一個併發度 Watermark 推進的進度，從而去統計 GlobalWatermark 的進展。根據 GlobalWatermark 的進展來計算出當中有哪些併發度的 Watermark 計算過快的問題，從而透過 GlobalAgg 下發給 Kafka Source 控制資訊，Kafka Source 有些併發度過快的情況下，它的整個分割槽推進就降低速度。這樣，在 HDFS Sink 模組，在同時間片上收到的資料記錄的整個 Event time 基本上有序的，最終落到 HDFS 還會在檔名上去標識它相應的分割槽以及相應的時間片範圍。最後在 HDFS 分割槽目錄下，就可以實現資料分割槽的有序目錄。

9。2 離線-分割槽增量消費

資料在 HDFS 增量有序之後，實現了 HDFStreamingSource，它會針對檔案做 Fecher 分割槽，針對每個檔案都有 Fecher 的執行緒，且每個 Fecher 執行緒會統計每一個檔案。它 offset 處理了遊標的進度，會將狀態根據 Checkpoint 的過程，將它更新到的 State 當中。

這樣就可以實現整個檔案消費的有序推進。在回溯歷史資料的時候，離線作業就會涉及到整個作業的停止。實際是在整個 FileFetcher 的模組當中去引入一個分割槽結束的標識，且會在每一個執行緒去統計每一個分割槽的時候，去感應它分割槽的結束，分割槽結束後的狀態最後彙總到的 cancellationManager，並進一步會彙總到 Job Manager 去更新全域性分割槽的進度，當全域性所有的分割槽都到了末尾的遊標時候，會將整個 Flink 作業進行 cancel 關閉掉。

9。3 離線 - Snapshot 維表

前面講到整個離線資料，其實資料都在 hive 上，hive 的 HDFS 表資料的整個表字段資訊會非常的多，但實際做離線特徵的時候，需要的資訊其實是很少的，因此需要在 hive 的過程先做離線欄位裁剪，將一張 ODS 的表清洗成 DW 的表，DW 的表會最後透過 Flink 執行 Job，內部會有個 reload 的 scheduler，它會定期的去根據資料當前推進的 Watermark 的分割槽，去拉取在 hive 當中每一個分割槽對應的表資訊。透過去下載某 HDFS 的 hive 目錄當中的一些資料，最後會在整個記憶體當中 reload 成 Rocksdb 的檔案，Rocksdb 其實就是最後用來提供維表 KV 查詢的元件。

元件裡面會包含多個 Rocksdb 的 build 構建過程，主要是取決於整個資料流動的過程當中的 Eventtime，如果發現 Eventtime 推進已經快到小時分割槽結束的末尾時候，會透過懶載入的模式去主動 reload，構建下一個小時 Rocksdb 的分割槽，透過這種方式，來切換整個 Rocksdb 的讀取。

10. 實驗流批一體

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

來看一下近線，其實 Label join 用的是 Kafka 的一條點選流以及展現流，到了整個離線的計算鏈路，則用的一條 HDFS 點選的目錄和 HDFS 展現目錄。特徵資料處理也是一樣的，實時用的是 Kafka 的播放資料，以及 Hbase 的一些稿件資料。對於離線來說，用的是 hive 的稿件資料，以及 hive 的播放資料。除了整個離線和近線的流批打通，還將整個近線產生的實時的資料效果彙總到 OLAP 引擎上，透過 superset 來提供整個實時的指標視覺化。其實從圖可以看出完整的複雜流批一體的計算鏈路，當中包含的計算節點是非常的複雜和龐多的。

11. 實驗協作 - 挑戰

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

12. 引入 AIFlow

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

右側其實是整個 AIFlow 完整鏈路的DAG檢視，可以看出整個節點，其實它支援的型別是沒有任何限制的，可以是流式節點，也可以是離線節點。此外的話，整個節點與節點之間通訊的邊是可以支援資料驅動以及事件驅動的。引入 AIFlow 的好處主要在於，AIFlow 提供基於 Python 語義來方便去定義完整的 AIFlow 的工作流，同時還包括整個工作流的進度的排程。

在節點的邊上，相比原生的業界的一些 Flow 方案，他還支援基於事件驅動的整個機制。好處是可以幫助在兩個 Flink 作業之間，透過 Flink 當中 watermark 處理資料分割槽的進度去下發一條事件驅動的訊息來拉起下一個離線或者實時的作業。

此外還支援周邊的一些配套服務，包括通知的一些訊息模組服務，還有元資料的服務，以及在 AI 領域一些模型中心的服務。

13. Python 定義 Flow

來看一下基於 AIFlow 是如何最終定義成 Python 的工作流。右邊的檢視是一個線上專案的完整工作流的定義。第一、是整個是 Spark job 的定義，當中透過配置 dependence 來描述整個下游的依賴關係，它會下發一條事件驅動的訊息來拉起下面的 Flink 流式作業。流式作業也同樣可以透過訊息驅動的方式來拉起下面的 Spark 作業。整個語義的定義非常的簡單，只需要四個步驟，配置每節點的 confg 的資訊，以及定義每節點的 operation 的行為，還有它的 dependency 的依賴，最後去執行整個 flow 的拓撲檢視。

14. 基於事件驅動流批

接下來看一下完整的流批排程的驅動機制，下圖右側是完整的三個工作節點的驅動檢視。第一個是從 Source 到 SQL 到 Sink。引入的黃色方框是擴充套件的 supervisor，他可以收集全域性的 watermark 進度。當整個流式作業發現 watermark 可以推進到下一個小時的分割槽的時候，它會下發一條訊息，去給到 NotifyService。NotifyService 拿到這條訊息之後，它會去下發給到下一個作業，下一個作業主要會在整個 Flink 的 DAG 當中去引入 flow 的 operator，operator 在沒有收到上個作業下發了訊息之前，它會堵塞整個作業的執行。直到收到訊息驅動之後，就代表上游其實上一個小時分割槽已經完成了，這時下個 flow 節點就可以驅動拉起來運作。同樣，下個工作流節點也引入了 GlobalWatermark Collector 的模組來彙總收集它的處理的進度。當上一個小時分割槽完成之後，它也會下發一條訊息到 NotifyService，NotifyService 會將這條訊息去驅動呼叫 AIScheduler 的模組，從而去拉起 spark 離線作業來做 spark 離線的收尾。從裡你們可以看出，整個鏈路其實是支援批到批，批到流以及流到流，以及流到批的四個場景。

15. 實時 AI 全鏈路的雛形

在流和批的整個 flow 定義和排程的基礎上，在 2020 年初步構建出來了實時 AI 全鏈路的雛形，核心是面向實驗。演算法同學也可以基於 SQL 來開發的 Node 的節點，Python 是可以定義完整的 DAG 工作流。監控，告警以及運維是一體化的。

同時，支援從離線到實時的打通，從資料處理到模型訓練，從模型訓練到實驗效果的打通，以及面向端到端的打通。右側是整個近線實驗的鏈路。下面是將整個實驗鏈路產出的物料資料提供給線上的預測訓練的服務。整體會有三個方面的配套：

一是基礎的一些平臺功能，包括實驗管理，模型管理，特徵管理等等；

其次也包括整個 AIFlow 底層的一些 service 的服務；

再有是一些平臺級的 metadata 的元資料服務。

四、未來的一些展望

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

第一是資料湖的方向上，會集中在 ODS 到 DW 層的一些增量計算場景，以及 DW 到 ADS 層的一些場景的突破，核心會結合 Flink 加 Iceberg 以及 HUDI 來作為該方向的落地。

在實時 AI 平臺上，會進一步去面向實驗來提供一套實時的 AI 協作平臺，核心是希望打造高效，能夠提煉簡化演算法人員的工程平臺。

原文連結

本文為阿里雲原創內容，未經允許不得轉載。