分享嘉賓：張楊，B 站資深開發工程師
導讀：整個機器學習的過程，從資料上報、到特征計算、到模型訓練、再到線上部署、最終效果評估，整個流程非常冗長。在 b 站，多個團隊都會搭建自己的機器學習鍊路，來完成各自的機器學習需求，工程效率和資料品質都難以保證。于是我們基于 Flink 社群的 aiflow 項目，建構了整套機器學習的标準工作流平台，加速機器學習流程建構，提升多個場景的資料實效和準确性。本次分享将介紹 b 站的機器學習工作流平台 ultron 在 b 站多個機器學習場景上的應用。
目錄：
1、機器學習實時化
2、Flink 在 B 站機器學習的使用
3、機器學習工作流平台建構
4、未來規劃
GitHub 位址
https://github.com/apache/flink歡迎大家給 Flink 點贊送 star~
一、機器學習實時化
首先講下機器學習的實時化，主要是分為三部分：
- 第一是樣本的實時化。傳統的機器學習，樣本全部都是 t+1，也就是說，今天模型用的是昨天的訓練資料，每天早上使用昨天的全天資料訓練一次模型；
- 第二是特征的實時化。以前的特征也基本都是 t+1，這樣就會帶來一些推薦不準确的問題。比如，今天我看了很多新的視訊，但給我推薦的卻還是一些昨天或者更久之前看到的内容；
- 第三就是模型訓練的實時化。我們有了樣本的實時化和特征的實時化之後，模型訓練也是完全可以做到線上訓練實時化的，能帶來更實時的推薦效果。
傳統離線鍊路
上圖是傳統的離線鍊路圖，首先是 APP 産生日志或者服務端産生 log，整個資料會通過資料管道落到 HDFS 上，然後每天 t+1 做一些特征生成和模型訓練，特征生成會放到特征存儲裡面，可能是 redis 或者一些其他的 kv 存儲，再給到上面的 inference 線上服務。
傳統離線鍊路的不足
那它有什麼問題呢？
- 第一是 t+1 資料模型的特征時效性都很低，很難做到特别高時效性的更新；
- 第二是整個模型訓練或者一些特征生産的過程中，每天都要用天級的資料，整個訓練或者特征生産的時間非常長，對叢集的算力要求非常高。
實時鍊路
上圖我們進行優化之後整個實時鍊路的過程，紅叉的部分是被去掉的。整個資料上報後通過 pipeline 直接落到實時的 kafka，之後會做一個實時特征的生成，還有實時樣本的生成，特征結果會寫到 feature store 裡面去，樣本的生成也需要從 feature store 裡面去讀取一些特征。
生成完樣本之後我們直接進行實時訓練。整個右邊的那個很長的鍊路已經去掉了，但是離線特征的部分我們還是儲存了。因為針對一些特殊特征我們還是要做一些離線計算，比如一些特别複雜不好實時化的或者沒有實時化需求的。
二、Flink 在 b 站機器學習的使用
下面講下我們是怎麼做到實時樣本、實時特征和實時效果評估的。
- 第一個是實時樣本。Flink 目前托管 b 站所有推薦業務樣本資料生産流程；
-
第二個是實時特征。目前相當一部分特征都使用了 Flink 進行實時計算，時效性非常高。有很多特征是使用離線 + 實時組合的方式得出結果，曆史資料用離線算，實時資料用 Flink，讀取特征的時候就用拼接。
但是，這兩套計算邏輯有的時候不能複用，是以我們也在嘗試使用 Flink 做批流一體，将特征的定義全部用 Flink 來做，根據業務需要，實時算或者離線算，底層的計算引擎全部是 Flink；
- 第三是實時效果的一個評估，我們使用了 Flink+olap 來打通整個實時計算 + 實時分析鍊路，進行最終的模型效果評估。
實時樣本生成
上圖是目前實時樣本的生成，是針對整個推薦業務鍊路的。日志資料落入 kafka 後，首先我們做一個 Flink 的 label-join，把點選和展現進行拼接。結果繼續落入 kafka 後，再接一個 Flink 任務進行特征 join，特征 join 會拼接多個特征，有些特征是公域特征，有些是業務方的私域特征。特征的來源比較多樣，有離線也有實時。特征全部補全之後，就會生成一個 instance 樣本資料落到 kafka，給後面的訓練模型使用。
實時特征生成
上圖是實時特征的生成，這邊列的是一個比較複雜的特征的過程，整個計算流程涉及到了 5 個任務。第一個任務是離線任務，後面有 4 個 Flink 任務，一系列複雜計算後生成的一個特征落到 kafka 裡面，再寫入 feature-store，然後被線上預測或者實時訓練所用到。
實時效果評估
上圖是實時效果的評估，推薦算法關注的一個非常核心的名額就是 ctr 點選率，做完 label-join 之後，就可以算出 ctr 資料了，除了進行下一步的樣本生成之外，同時會導一份資料到 clickhouse 裡面，報表系統對接後就可以看到非常實時的效果。資料本身會帶上實驗标簽，在 clickhouse 裡面可以根據标簽進行實驗區分，看出對應的實驗效果。
三、機器學習工作流平台建構
痛點
- 機器學習的整個鍊路裡面有樣本生成、特征生成、訓練、預測、效果評估，每個部分都要配置開發很多任務，一個模型的上線最終需要橫跨多個任務，鍊路非常長。
- 新的算法同學很難去了解這個複雜鍊路的全貌，學習成本極高。
- 整個鍊路的改動牽一發而動全身，非常容易出故障。
- 計算層用到多個引擎，批流混用，語義很難保持一緻，同樣的邏輯要開發兩套，保持沒有 gap 也很困難。
- 整個實時化成本門檻也比較高，需要有很強的實時離線能力，很多小的業務團隊在沒有平台支援下難以完成。
上圖是一個模型從資料準備到訓練的大概過程，中間涉及到了七八個節點，那我們能不能在一個平台上完成所有的流程操作？我們為什麼要用 Flink？是因為我們團隊實時計算平台是基于 Flink 來做的，我們也看到了 Flink 在批流一體上的潛力以及在實時模型訓練和部署上一些未來發展路徑。
引入 Aiflow
Aiflow 是阿裡的 Flink 生态團隊開源的一套機器學習工作流平台，專注于流程和整個機器學習鍊路的标準化。去年八、九月份，我們在和他們接觸後，引入了這樣一套系統，一起共建完善，并開始逐漸在 b 站落地。它把整個機器學習抽象成圖上的 example、transform 、Train、validation、inference 這些過程。在項目架構上非常核心的能力排程就是支援流批混合依賴，中繼資料層支援模型管理，非常友善的進行模型的疊代更新。我們基于此搭建了我們的機器學習工作流平台。
平台特性
接下來講一下平台特性：
- 第一是使用 Python 定義工作流。在 ai 方向，大家用 Python 還是比較多的，我們也參考了一些外部的，像 Netflix 也是使用 Python 來定義這種機器學習的工作流。
- 第二是支援批流任務混合依賴。在一個完整鍊路裡面，涉及到的實時離線過程都可以加入到裡面，并且批流任務之間可以通過信号就行互相依賴。
- 第三是支援一鍵克隆整個實驗過程。從原始 log 到最終整個實驗拉起訓練這塊，我們是希望能夠一鍵整體鍊路克隆，快速拉起一個全新的實驗鍊路。
- 第四是一些性能方面的優化，支援資源共享。
- 第五是支援特征回溯批流一體。很多特征的冷啟動需要計算曆史很長時間的資料，專門為冷啟動寫一套離線特征計算邏輯成本非常高，而且很難和實時特征計算結果對齊，我們支援直接在實時鍊路上來回溯離線特征。
基本架構
上圖是基本架構，最上面是業務，最下面是引擎。目前支援的引擎也比較多：Flink、spark、Hive、kafka、Hbase、Redis。其中有計算引擎，也有存儲引擎。以 aiflow 作為中間的工作流程管理，Flink 作為核心的計算引擎，來設計整個工流平台。
工作流描述
整個工作流是用 Python 來描述的，在 python 裡面使用者隻需要定義計算節點和資源節點，以及這些節點之間的依賴關系即可，文法有點像排程架構 airflow。
依賴關系定義
批流的依賴關系主要有 4 種：流到批，流到流，批到流，批到批。基本可以滿足目前我們業務上的所有需求。
資源共享
資源共享主要是用來做性能方面，因為很多時候一個機器的學習鍊路非常長，比如剛剛那個圖裡面我經常改動的可能隻有五六個節點，當我想重新拉起整個實驗流程，把整個圖克隆一遍，中間我隻需要改動其中的部分節點或者大部分節點，上遊節點是可以做資料共享的。
這個是技術上的實作，克隆之後對共享節點做了一個狀态追蹤。
實時訓練
上圖是實時訓練的過程。特征穿越是一個非常常見的問題，多個計算任務的進度不一緻時就會發生。在工作流平台裡面，我們定義好各個節點的依賴關系即可，一旦節點之間發生了依賴，處理進度就會進行同步，通俗來說就是快的等慢的，避免特征穿越。在 Flink 裡面我們是使用 watermark 來定義處理進度。
特征回溯
上圖是特征回溯的過程，我們使用實時鍊路，直接去回溯它曆史資料。離線和實時資料畢竟不同，這中間有很多問題需要解決，是以也用到了 spark，後面這塊我們會改成 Flink。
特征回溯的問題
特征回溯有幾個比較大的問題：
- 第一是如何保證資料的順序性。實時資料有個隐含的語義就是資料是順序進來的，生産出來立馬處理，天然有一定的順序性。但是離線的 HDFS 不是，HDFS 是有分區的，分區内的資料完全亂序，實際業務裡面大量計算過程是依賴時序的，如何解決離線資料的亂序是一個很大的問題。
- 第二是如何保證特征和樣本版本的一緻性。比如有兩條鍊路，一條是特征的生産，一條是樣本生産，樣本生産依賴特征生産，如何保證它們之間版本的一緻性，沒有穿越？
- 第三就是如何保證明時鍊路和回溯鍊路計算邏輯的一緻？這個問題其實對我們來說不用擔心，我們是直接在實時鍊路上回溯離線資料。
- 第四是一些性能方面的問題，怎麼快速得算完大量的曆史資料。
解決方案
以下是第一、第二個問題的解決方案：
- 第一個問題。為了資料的順序性，我們 HDFS 的離線資料進行 kafka 化處理，這裡不是把它灌到 kafka 裡面去，而是模拟 kafka 的資料架構，分區并且分區内有序，我們把 HDFS 資料也處理成類似的架構，模拟成邏輯上的分區，并且邏輯分區内有序，Flink 讀取的 hdfssource 也進行了對應的開發支援這種模拟的資料架構。這塊的模拟計算目前是使用 spark 做的，後面我們會改成 Flink。
- 第二個問題分為兩部分：
- 實時特征部分的解決依賴于 Hbase 存儲，Hbase 支援根據版本查詢。特征計算完後直接按照版本寫入 Hbase，樣本生成的時候去查 Hbase 帶上對應的版本号即可，這裡面的版本通常是資料時間。
- 離線特征部分，因為不需要重新計算了，離線存儲 hdfs 都有，但是不支援點查，這塊進行 kv 化處理就好，為了性能我們做了異步預加載。
異步預加載的過程如圖。
四、未來規劃
接下來介紹下我們後面規劃。
- 一個是資料品質保證。現在整個鍊路越來越長，可能有 10 個節點、 20 個節點，那怎麼在整個鍊路出問題的時候快速發現問題點。這裡我們是想針對節點集來做 dpc，對每個節點我們可以自定義一些資料品質校驗規則，資料通過旁路到統一的 dqc-center 進行規則運算告警。
- 第二是全鍊路的 exactly once，工作流節點之間如何保證精确一緻，這塊目前還沒有想清楚。
- 第三是我們會在工作流裡面加入模型訓練和部署的節點。訓練和部署可以是連接配接到别的平台，也可能是 Flink 本身支援的訓練模型和部署服務。
嘉賓介紹：張楊，17 年入職 b 站，從事大資料方面工作。
更多 Flink 相關技術交流，可掃碼加入社群釘釘大群～
活動推薦
阿裡雲基于 Apache Flink 建構的企業級産品-實時計算Flink版現開啟活動：
99元試用
實時計算Flink版(包年包月、10CU）即有機會獲得 Flink 獨家定制T恤；另包3個月及以上還有85折優惠！
了解活動詳情：
https://www.aliyun.com/product/bigdata/sc