微博基于Flink的機(jī)器學(xué)習(xí)實(shí)踐

導(dǎo)讀：微博作為國(guó)內(nèi)比較主流的社交媒體平臺(tái)，目前擁有2.22億日活用戶和5.16億月活用戶。如何為用戶實(shí)時(shí)推薦優(yōu)質(zhì)內(nèi)容，背后離不開微博的大規(guī)模機(jī)器學(xué)習(xí)平臺(tái)。本文由微博機(jī)器學(xué)習(xí)研發(fā)中心高級(jí)算法工程師于茜老師分享，主要內(nèi)容包含以下四部分：

• 關(guān)于微博

• 微博機(jī)器學(xué)習(xí)平臺(tái) ( WML ) 總覽

• Flink在WML中的應(yīng)用

• 使用Flink的下一步計(jì)劃

微博2008年上線，是目前國(guó)內(nèi)比較主流的社交媒體平臺(tái)，擁有2.22億日活用戶和5.16億月活用戶，為用戶提供在線創(chuàng)作、分享和發(fā)現(xiàn)優(yōu)質(zhì)內(nèi)容的服務(wù)；目前微博的大規(guī)模機(jī)器學(xué)習(xí)平臺(tái)可以支持千億參數(shù)和百萬(wàn)QPS。

接下來(lái)介紹一下微博機(jī)器學(xué)習(xí)平臺(tái)，即WML的總覽；機(jī)器學(xué)習(xí)平臺(tái) ( WML ) 為CTR、多媒體等各類機(jī)器學(xué)習(xí)和深度學(xué)習(xí)算法提供從樣本處理、模型訓(xùn)練、服務(wù)部署到模型預(yù)估的一站式服務(wù)。

1. 總覽

上方是WML的一個(gè)整體架構(gòu)圖，共分為六層，從下至上依次介紹：

• 集群層：包含離線計(jì)算集群、在線計(jì)算集群和高性能計(jì)算集群；

• 調(diào)度層：包含自研的WeiBox ( 提供使用通用的接口將任務(wù)提交到不同集群的能力 )、Weiflow ( 提供將任務(wù)間的依賴關(guān)系處理好、組成DAG工作流的能力 )，以及常見的調(diào)度引擎Yarn和K8s；

• 計(jì)算平臺(tái)層：包含自研的WeiLearn ( 提供給用戶在該平臺(tái)做業(yè)務(wù)開發(fā)的能力 )，以及Hadoop/Spark離線計(jì)算平臺(tái)、Flink/Storm在線計(jì)算平臺(tái)和Tensorflow機(jī)器學(xué)習(xí)平臺(tái)；

• 模型訓(xùn)練層：目前支持LR、GBDT、FM/FFM、CF/MF、DNN/RNN等主流的算法；

• 在線推理層：包含自研的WeiServing和WeiPS；

• 業(yè)務(wù)應(yīng)用層：主要應(yīng)用場(chǎng)景是特征生成、樣本服務(wù)、在線訓(xùn)練和在線推理；

• 右邊是自定義的一些概念，樣本庫(kù)、模型庫(kù)、服務(wù)庫(kù)以及兩個(gè)任務(wù)提交方式WeiClient ( CLI方式提交 )、WAIC UI ( 界面操作 )。

2. 開發(fā)模式

接下來(lái)介紹一下開發(fā)模式，有兩層DAG的設(shè)計(jì)：

• 內(nèi)層，WeiLearn層里面可以重寫離線的Input、Process和Output方法以及實(shí)時(shí)的Source、Process和Sink方法，用戶自己開發(fā)一個(gè)UDF來(lái)實(shí)現(xiàn)自己的業(yè)務(wù)邏輯；內(nèi)層的每一個(gè)DAG都會(huì)組成一個(gè)Task。

• 外層，即第二層DAG層，WeiFlow層里面將WeiLearn中產(chǎn)生的Task的依賴關(guān)系組成一個(gè)集群內(nèi)或者跨集群的WorkFlow，然后運(yùn)行計(jì)算。

3. CTR模型

介紹一下CTR模型在微博迭代的情況，經(jīng)過(guò)幾年的研究和探索，目前支撐的參數(shù)規(guī)模達(dá)千億級(jí)，服務(wù)峰值達(dá)百萬(wàn)QPS，模型更新的周期大概在10分鐘左右；現(xiàn)在是Weilearn6.0版本，可以看到WeiLearn在不斷完善更新自己的算法：

• 1.0版本僅支持LR離線學(xué)習(xí)

• 2.0版本支持LR/GBDT/LR+GBDT離線學(xué)習(xí)

• 3.0版本支持LR/GBDT/LR+GBDT離線學(xué)習(xí)以及Wide&Deep的深度學(xué)習(xí)

• 4.0版本支持LR/GBDTLR+GBDT/FM/MF離線學(xué)習(xí)以及Wide&Deep的深度學(xué)習(xí)

• 5.0版本支持Online FM/FFM在線學(xué)習(xí)，LR/GBDT/LR+GBDT/FM/MF離線學(xué)習(xí)以及Wide&Deep/DeepFM/DSSM的深度學(xué)習(xí)

• 6.0版本更新了Online DNN模型，加強(qiáng)在線機(jī)器學(xué)習(xí)模型的表達(dá)能力

下面介紹Flink在微博機(jī)器學(xué)習(xí)平臺(tái)WML中的架構(gòu)

1. 概覽

上圖為實(shí)時(shí)計(jì)算平臺(tái)的整體情況，接下來(lái)詳細(xì)介紹一下各模塊：

• 基礎(chǔ)架構(gòu)層：包含Storm集群、Flink集群、Flume以及用于監(jiān)控系統(tǒng)運(yùn)行的Grafana。

• 計(jì)算層：主要是對(duì)Pig和Flink的進(jìn)一步封裝，包含WeiPig + WeiStream和WeiLearn + WeiFlink；左側(cè)為實(shí)時(shí)數(shù)據(jù)源，包含實(shí)時(shí)消息隊(duì)列、Redis、Kafka；一些歷史數(shù)據(jù)會(huì)存到右側(cè)的HDFS中。

• 應(yīng)用層：目前這套平臺(tái)主要應(yīng)用于多媒體特征生成、內(nèi)容去重、數(shù)據(jù)同步、實(shí)時(shí)特征生成、樣本服務(wù)以及在線訓(xùn)練。

• 業(yè)務(wù)層：支撐了目前微博主要的幾個(gè)業(yè)務(wù)，包含熱門微博、關(guān)系流、視頻推薦、內(nèi)容監(jiān)控和圖片推薦。

接下來(lái)看一下Flink在ETL的Pipeline中的概覽：之前是有兩個(gè)Pipeline，一個(gè)為在線的，以前是使用Storm進(jìn)行的處理，目前正在往Flink遷移，兩套現(xiàn)在處于并行狀態(tài)，處理流程是從消息隊(duì)列中獲取數(shù)據(jù)進(jìn)行處理，然后給到在線訓(xùn)練模塊 ( Flink和Spark Streaming并行 )，最后提供模型服務(wù)給推薦系統(tǒng)調(diào)用；一個(gè)為離線的，和在線類似，首先寫入到HDFS交給Hive或Spark進(jìn)行處理，再次落到HDFS中交給離線訓(xùn)練使用，最后提供模型服務(wù)給推薦系統(tǒng)調(diào)用。因?yàn)橛袃深怑TL的Pipeline，使用不同的框架，需要維護(hù)兩套代碼，維護(hù)成本較高。

目前做的就是將兩套融合成一套，進(jìn)行批流統(tǒng)一的處理，此處可能會(huì)用到FlinkSQL，然后將ETL后的數(shù)據(jù)輸出到實(shí)時(shí)消息隊(duì)列或者HDFS中，交給在線和離線模型訓(xùn)練，最后提供模型服務(wù)給推薦系統(tǒng)調(diào)用。

2. 樣本服務(wù)

介紹一下樣本生成服務(wù)，上圖為該服務(wù)的整體架構(gòu)圖，包含樣本數(shù)據(jù)的處理和計(jì)算等，除了一些生成的離線和實(shí)時(shí)數(shù)據(jù)外，還需要一些已經(jīng)生成好的特征的引用，通過(guò)普通計(jì)算、多流Join、深度學(xué)習(xí)等處理方式生成樣本，最后存儲(chǔ)到樣本庫(kù)中供模型訓(xùn)練來(lái)調(diào)用。

這個(gè)是樣本服務(wù)任務(wù)提交的方式，可以通過(guò)之前提到的WeiClient命令行方式提交，也可以通過(guò)WAIC UI方式指定樣本ID以及UDF的class name和要拼接的特征ID，通過(guò)一種統(tǒng)一的方式將作業(yè)提交到集群上；之后是通過(guò)Twinkle或VVP的方式提交到Flink集群，然后會(huì)對(duì)作業(yè)狀態(tài)進(jìn)行管理，通過(guò)Grafana進(jìn)行監(jiān)控和報(bào)警，將歷史作業(yè)信息存儲(chǔ)到HDFS中。

3. 多流Join

這是微博目前的一個(gè)主流場(chǎng)景，多數(shù)據(jù)流Join場(chǎng)景 ( 大部分是大于等于3 )：有N個(gè)數(shù)據(jù)源，通過(guò)過(guò)濾和映射的處理后按照Key進(jìn)行分發(fā)，在Joining Window中進(jìn)行join后 ( 此處后面會(huì)詳細(xì)講 )，會(huì)再進(jìn)行一次過(guò)濾和映射以及添加特征，最后輸出到樣本庫(kù)中。

接下來(lái)看一下剛剛講到的拼接窗口的實(shí)現(xiàn)方式，這是和業(yè)務(wù)比較相關(guān)的，對(duì)于CTR場(chǎng)景來(lái)說(shuō)日志有很多種 ( 多個(gè)行為日志 )，但是到達(dá)的時(shí)間并不完全一致，比如點(diǎn)擊這種行為日志可能會(huì)比曝光日志到的晚一些；這樣就會(huì)需要一個(gè)時(shí)間窗口，以10分鐘為例，如果某種日志先到了，就會(huì)將對(duì)應(yīng)的key和value存儲(chǔ)到State中，狀態(tài)存儲(chǔ)這塊是基于RocksDB和HDFS做的；經(jīng)過(guò)這個(gè)十分鐘窗口之后，拼接好的樣本數(shù)據(jù)會(huì)輸?shù)綄?shí)時(shí)流中；此處基于Flink做了一些優(yōu)化：

• 因?yàn)榇翱谑?0分鐘的，但是如果10分鐘內(nèi)日志數(shù)據(jù)已經(jīng)全部到達(dá)，就不同等到10分鐘窗口結(jié)束后再輸出去；所以自定義了樣本trigger觸發(fā)機(jī)制，樣本拼接成功后就可以立即輸出，這樣可以減少一些時(shí)延

• 樣本補(bǔ)償 PU loss；此處是基于Twitter在2019年發(fā)的一篇論文的實(shí)現(xiàn)方式，就是拿到正樣本之后，首先對(duì)正樣本做一個(gè)梯度下降的處理，另外可能之前有False Negative的樣本已經(jīng)發(fā)送出去了，那就需要之前的樣本進(jìn)行補(bǔ)償，所以需要對(duì)該樣本的負(fù)樣本做一個(gè)反向的梯度下降

• 另外在RocksDB做狀態(tài)存儲(chǔ)這部分，引用了Gemini與RocksDB作對(duì)比，Gemini的IO性能更好一些

• 拼接窗口時(shí)長(zhǎng)的控制是和業(yè)務(wù)場(chǎng)景比較相關(guān)的，日志到達(dá)的時(shí)間和具體的業(yè)務(wù)場(chǎng)景是有關(guān)系的，所以需要權(quán)衡時(shí)間窗口設(shè)置多長(zhǎng)時(shí)間才能滿足拼接成功率的預(yù)期，這塊需要大量的離線計(jì)算和A/B Test來(lái)共同決定。

4. 多媒體特征生成

介紹一下Flink在多媒體特征生成場(chǎng)景的應(yīng)用，此處主要是依賴離線計(jì)算的深度學(xué)習(xí)模型，因此整體的模型訓(xùn)練走的是離線的Pipeline，將數(shù)據(jù)在離線的GPU集群進(jìn)行分布式的模型訓(xùn)練，然后將模型部署到GPU上面供在線推理的時(shí)候調(diào)用；在線推理模塊接收到圖片流、文本流和視頻流這些實(shí)時(shí)數(shù)據(jù)之后，首先會(huì)通過(guò)RPC調(diào)用GPU上的模型，然后將多媒體特征結(jié)果寫入到數(shù)據(jù)中臺(tái)，由業(yè)務(wù)方去讀取結(jié)果來(lái)使用，因?yàn)檫@塊是一個(gè)實(shí)時(shí)的任務(wù)作業(yè)，服務(wù)穩(wěn)定性需要一定的保障 ( 4個(gè)9的成功率、秒級(jí)延遲、配置化開發(fā)模式 )，下面會(huì)對(duì)服務(wù)保障做詳細(xì)介紹。

針對(duì)實(shí)時(shí)任務(wù)的服務(wù)保障做了如下的工作：

• 全鏈路監(jiān)控報(bào)警&Case追蹤，針對(duì)模型服務(wù)到RPC的情況、模型關(guān)鍵指標(biāo)以及樣本情況整體是有一個(gè)全流程的監(jiān)控

• 設(shè)置消息機(jī)制是At least once，每條消息至少要被處理一次，這樣可以保障每條數(shù)據(jù)結(jié)果都能寫到特征工程中

• 任何一個(gè)部分出現(xiàn)問(wèn)題都會(huì)實(shí)現(xiàn)自動(dòng)重啟

• 重啟時(shí)可以從checkpoints中恢復(fù)數(shù)據(jù)和State，可以避免一些重復(fù)計(jì)算，也是為了減少一些延時(shí)

• 所有實(shí)時(shí)任務(wù)都會(huì)起一個(gè)重試的任務(wù)，這樣在主流程中寫入失敗，會(huì)再次寫入到重試隊(duì)列中再進(jìn)行一次重試的寫入，這樣保障數(shù)據(jù)會(huì)被計(jì)算兩次；如果最終還是寫入失敗，就會(huì)記錄到對(duì)賬離線系統(tǒng)中，這樣可以看到哪些數(shù)據(jù)是寫入失敗的，可以手動(dòng)恢復(fù)一下。

最后分享一下使用Fllink的下一步計(jì)劃：

1. 實(shí)時(shí)數(shù)倉(cāng)

目前已經(jīng)通過(guò)Flink SQL的方式實(shí)現(xiàn)了開發(fā)，但是實(shí)時(shí)和離線表的注冊(cè)還有元數(shù)據(jù)存儲(chǔ)是有一定差異的，希望可以抽象出一層API用統(tǒng)一的方式來(lái)進(jìn)行實(shí)時(shí)和離線表的注冊(cè)以及元數(shù)據(jù)的存儲(chǔ)。

2. 基于Flink的DL

我們希望可以將離線的深度學(xué)習(xí)完全遷移到在線深度學(xué)習(xí)來(lái)做，這樣的話就需要用到TensorFlow on Flink，這樣就可以保證不管是模型訓(xùn)練還是在線推理都可以使用同樣一套框架去完成，這樣就需要把離線訓(xùn)練的全量模型也可以通過(guò)實(shí)時(shí)樣本進(jìn)行增量訓(xùn)練的一些校正，后面的步驟和之前基本上是保持一致的，這樣就可以將離線深度學(xué)習(xí)的這條Pipeline優(yōu)化一些。

本次的分享就到這里，謝謝大家。

嘉賓介紹：

于茜，微博機(jī)器學(xué)習(xí)研發(fā)中心高級(jí)算法工程師。多年來(lái)致力于使用 Flink 構(gòu)建實(shí)時(shí)數(shù)據(jù)處理和在線機(jī)器學(xué)習(xí)框架，有豐富的社交媒體應(yīng)用推薦系統(tǒng)的開發(fā)經(jīng)驗(yàn)。

特別推薦一個(gè)分享架構(gòu)+算法的優(yōu)質(zhì)內(nèi)容，還沒關(guān)注的小伙伴，可以長(zhǎng)按關(guān)注一下：

長(zhǎng)按訂閱更多精彩▼
如有收獲，點(diǎn)個(gè)在看，誠(chéng)摯感謝