实时数仓入门训练营：基于 Apache Flink + Hologres 的实时推荐系统架构解析
本文整理自直播《基于 Apache Flink + Hologres 的實(shí)時(shí)推薦系統(tǒng)架構(gòu)解析-秦江杰》
視頻鏈接:https://c.tb.cn/F3.0d98Xr
內(nèi)容簡要：
一、實(shí)時(shí)推薦系統(tǒng)原理
二、實(shí)時(shí)推薦系統(tǒng)架構(gòu)
三、基于 Apache Flink + Hologres 的實(shí)時(shí)推薦系統(tǒng)關(guān)鍵技術(shù)
實(shí)時(shí)推薦系統(tǒng)原理
（一）靜態(tài)推薦系統(tǒng)
在介紹實(shí)時(shí)推薦系統(tǒng)之前，先看一下靜態(tài)推薦系統(tǒng)是什么樣子的。
上方是一個(gè)非常經(jīng)典的靜態(tài)推薦系統(tǒng)的架構(gòu)圖。前端會(huì)有很多用戶端的應(yīng)用，這些用戶會(huì)產(chǎn)生大量用戶的行為日志，然后放到一個(gè)消息隊(duì)列里面，進(jìn)入ETL。接著通過離線系統(tǒng)去做一些特征生成和模型訓(xùn)練，最后把模型和特征推到線上系統(tǒng)中，通過在線的服務(wù)就可以去調(diào)用在線推理服務(wù)去獲得推薦結(jié)果。
這就是一個(gè)非常經(jīng)典的靜態(tài)推薦系統(tǒng)運(yùn)作流程，下面我們舉一個(gè)具體的例子來看靜態(tài)推薦系統(tǒng)到底是怎么樣工作的。
如上圖所示，比如在線用戶的行為日志可能是一些用戶的瀏覽和廣告點(diǎn)擊的日志，推薦系統(tǒng)的目的是為了幫用戶推薦廣告，那么在日志里面可以看到以下用戶行為：
用戶1和用戶2都看了PageID 200和一些其他的頁面，然后用戶1看了PageID 200并且點(diǎn)了廣告2002，那么在用戶日志里面通過ETL可以把這樣的一系列行為給歸納出來，然后送到模型訓(xùn)練里面去訓(xùn)練模型。在訓(xùn)練模型的過程當(dāng)中我們會(huì)用到一些特征，在這個(gè)情況下我們可以發(fā)現(xiàn)用戶1和用戶2都是中國的男性用戶，這可能是用戶維度的一個(gè)特征。
在這種情況下，我們從日志里面看到的結(jié)果是用戶在看了PageID 100后點(diǎn)了廣告2002，并且兩個(gè)用戶都是中國的男性用戶。因此，我們的模型就有可能學(xué)到當(dāng)中國的男性用戶來看PageID 100的時(shí)候，應(yīng)該要給他展示廣告2002，這個(gè)行為會(huì)被訓(xùn)練到模型里面去。這個(gè)時(shí)候我們會(huì)把一些用戶的離線特征都推到特征庫，然后把這個(gè)模型也推到線上去。
假設(shè)這里有一個(gè)用戶ID4，他正好是中國的男性用戶，這個(gè)特征就會(huì)被推進(jìn)特征庫，那模型也被推到線上。如果用戶4來訪問的時(shí)候看PageID 100，推理服務(wù)會(huì)先去看用戶ID4的特征，然后根據(jù)他是一個(gè)中國的男性用戶，通過訓(xùn)練的模型，系統(tǒng)就會(huì)給他推廣告2002，這是一個(gè)靜態(tài)推薦系統(tǒng)基本的工作原理。
在這種情況下，如果發(fā)生一些變化的時(shí)候，我們來看一下靜態(tài)推薦系統(tǒng)是不是能夠繼續(xù)很好地工作？
假使說今天訓(xùn)練了用戶1和用戶2的特征模型，到第二天發(fā)現(xiàn)用戶4產(chǎn)生了行為，根據(jù)模型里面的內(nèi)容，模型會(huì)認(rèn)為用戶4是中國的男性用戶和用戶1、用戶2行為一致，所以需要給他推的應(yīng)該是中國男性用戶的行為。但這個(gè)時(shí)候我們發(fā)現(xiàn)用戶4的行為其實(shí)跟用戶3更像，而不是跟用戶1和用戶2更像。
在這種情況下，由于模型和特征都是靜態(tài)的，所以為了讓用戶4能夠跟用戶3得到的行為更像，需要去重新訓(xùn)練模型，這會(huì)導(dǎo)致預(yù)測的效果被延遲，因?yàn)樾枰匦掠?xùn)練用戶4，才能夠推薦出跟用戶3更像的一些行為。
所以在這種實(shí)際操作情況下，可以看到靜態(tài)推薦模型存在一些問題：
- 靜態(tài)生成模型和特征；
-
以分類模型為例，根據(jù)用戶的相似性進(jìn)行用戶分類，假設(shè)同類用戶有相似的興趣和行為
- 例如中國的男性用戶有類似行為。
- 一旦用戶被劃分為某個(gè)類別，則他將一直處于這個(gè)類別中，直到被新的模型訓(xùn)練重新分類。
這種情況下，比較難去做到很好的推薦，原因是：
- 用戶的行為非常多元化，無法劃分到某個(gè)固定類別
1）上午為父母采購保健品，中午為出差訂酒店，晚上給家人買衣服…
2）靜態(tài)系統(tǒng)無法準(zhǔn)確將用戶放到當(dāng)時(shí)當(dāng)刻正確的類別中。 - 某一類別用戶的行為相似，但是行為本身可能會(huì)發(fā)生變化
1）假設(shè)用戶“隨大流“，但是“大流”可能發(fā)生變化；
2）歷史數(shù)據(jù)看出來的“大流”可能無法準(zhǔn)確反映線上的真實(shí)情況。
（二）加入實(shí)時(shí)特征工程的推薦系統(tǒng)
為了解決上述問題，可以加入動(dòng)態(tài)特征。那么動(dòng)態(tài)特征是什么樣的？舉個(gè)例子說明。
如上圖所示，我們以大流發(fā)生變化的動(dòng)態(tài)特征舉例。之前的模型推薦是如果中國的男性用戶訪問PageID 100，就給他推薦廣告2002，這是一個(gè)固定不變的行為。
在此基礎(chǔ)上做一些變化，當(dāng)進(jìn)行采樣實(shí)時(shí)特征的時(shí)候，這個(gè)實(shí)時(shí)特征是最近一段時(shí)間內(nèi)，即當(dāng)中國的男性用戶訪問PageID 100的時(shí)候，他們點(diǎn)擊最多的10個(gè)廣告。這個(gè)特征沒有辦法在離線的時(shí)候計(jì)算出來，因?yàn)樗且粋€(gè)線上實(shí)時(shí)發(fā)生的用戶行為。
那么在產(chǎn)生用戶行為之后可以做一件什么事情呢？可以在中國的男性用戶訪問PageID 100的時(shí)候，不單純給他推廣告2002，而是推最近這段時(shí)間中國男性用戶訪問PageID 100時(shí)候點(diǎn)擊最多的那些廣告。
這樣的情況下，如果中國男性用戶訪問PageID 100的時(shí)候，最近訪問最多的廣告是2001和2002。當(dāng)用戶ID來了，我們看到他是一個(gè)中國男性用戶，就有可能給他推薦廣告2001，而不是廣告2002了。
上述就是大流發(fā)生變化的一個(gè)例子。
同樣的道理，因?yàn)橄到y(tǒng)可以對(duì)用戶的實(shí)時(shí)特征進(jìn)行采樣，所以能更好地判斷用戶當(dāng)時(shí)當(dāng)刻的意圖。比方說，可以去看用戶最近一分鐘看了哪些頁面，瀏覽哪些商品，這樣的話可以實(shí)時(shí)判斷用戶當(dāng)時(shí)當(dāng)刻的想法，從而給他推薦一個(gè)更適合他當(dāng)下意圖的廣告。
這樣的推薦系統(tǒng)是不是就完全沒有問題呢？再看一個(gè)例子。
比方說剛才上文提到用戶1和用戶2都是中國男性用戶，之前假設(shè)他們的行為是類似的，在之前的歷史數(shù)據(jù)里面也印證了這一點(diǎn)。但是當(dāng)在線上真正看用戶行為的時(shí)候，可能會(huì)發(fā)生什么樣的情況？
可能發(fā)生用戶1和用戶2的行為產(chǎn)生分化，分化的原因可能有很多種，但不知道是什么原因。此時(shí)給用戶1和用戶2所推薦的東西可能就完全不一樣了，那是什么原因?qū)е路只?#xff1f;
舉個(gè)例子來說，如果用戶1來自上海，用戶2來自北京。某天北京有非常大的降溫，這個(gè)時(shí)候北京用戶2可能就開始搜索秋褲，但是上海當(dāng)天還是很熱，上海的用戶1在搜索服裝的時(shí)候，可能還是搜索一些夏裝。這個(gè)時(shí)候，中國的男性用戶里面，上海用戶1和北京用戶2的搜索行為就產(chǎn)生了一些變化。此時(shí)就需要給他們推薦不一樣的廣告，但是靜態(tài)的模型沒有辦法很好地做到這一點(diǎn)。
因?yàn)檫@個(gè)模型其實(shí)是一個(gè)靜態(tài)訓(xùn)練的模型，所以如果是一個(gè)分類模型的話，當(dāng)中能夠產(chǎn)生的類別其實(shí)是一個(gè)固定的類別，為了產(chǎn)生一個(gè)新的分類，就需要對(duì)模型重新進(jìn)行訓(xùn)練。由于模型訓(xùn)練是離線進(jìn)行的，所以可能這個(gè)訓(xùn)練的模型需要在第二天才能被更新，這樣就會(huì)對(duì)推薦效果產(chǎn)生影響。
- 通過增加動(dòng)態(tài) feature
1）實(shí)時(shí)跟蹤一類用戶的行為，貼合“大流”；
2）實(shí)時(shí)追蹤用戶的行為表現(xiàn)，了解用戶當(dāng)時(shí)當(dāng)刻的意圖，并將用戶劃分到更合適的類別中去。 - 但是當(dāng)模型的分類方式本身發(fā)生變化時(shí)，可能無法找到最合適的類別，需要重新訓(xùn)練模型增加分類。
例：新產(chǎn)品上線頻繁，業(yè)務(wù)高速成長，用戶行為的分布變化比較快。
當(dāng)遇到以上問題，需要把考慮的事情加入動(dòng)態(tài)的模型更新，動(dòng)態(tài)模型更新是怎么來做？其實(shí)是一樣的道理。
如上圖所示，除了把用戶的實(shí)時(shí)行為日志做ETL到離線的地方進(jìn)行Feature Generation以外，可能還要把用戶行為日志在線導(dǎo)出來，然后去做特征生成、樣本拼接，然后做進(jìn)線的模型訓(xùn)練。
這里的模型訓(xùn)練通常都是流式的訓(xùn)練，在一個(gè)基礎(chǔ)模型之上做增量的訓(xùn)練，來使模型更好地貼合當(dāng)時(shí)當(dāng)刻用戶行為的一些變化。在這種情況下，通過這種實(shí)時(shí)樣本的訓(xùn)練，可以讓這個(gè)模型產(chǎn)生新的分類，它會(huì)知道上海和北京用戶的行為可能是不一樣的。因此，當(dāng)用戶訪問PageID 100的時(shí)候，對(duì)于上海的用戶它可能會(huì)推薦廣告2002，北京的用戶可能推薦的就是廣告2011了。
在這樣的情況分化下，假設(shè)用戶4再過來的時(shí)候，系統(tǒng)會(huì)看他到底是上海的用戶還是北京的用戶，如果他是上海的用戶的話，還是會(huì)給他推薦廣告2002。
加入實(shí)時(shí)模型訓(xùn)練的推薦系統(tǒng)特點(diǎn)：
- 在動(dòng)態(tài)特征的基礎(chǔ)上，實(shí)時(shí)訓(xùn)練模型，使模型盡可能貼近此時(shí)此刻 用戶行為的分布；
- 緩解模型的退化。
實(shí)時(shí)推薦系統(tǒng)架構(gòu)
上面的例子是了解實(shí)時(shí)推薦系統(tǒng)的原理，它為什么會(huì)比一般的離線推薦系統(tǒng)做得更好。那么，如何通過Flink加上Hologres和一些其他系統(tǒng)/項(xiàng)目來搭建出這樣一套可用的實(shí)時(shí)推薦系統(tǒng)？
（一）經(jīng)典離線推薦系統(tǒng)架構(gòu)
首先來看一下上文提到的經(jīng)典離線推薦系統(tǒng)的架構(gòu)，如下所示。
這個(gè)架構(gòu)其實(shí)之前講的架構(gòu)一樣，只是增加了部分細(xì)節(jié)。
首先，通過消息隊(duì)列用來采集實(shí)時(shí)的用戶行為，這個(gè)消息隊(duì)列里面的實(shí)時(shí)用戶行為會(huì)被導(dǎo)入到一個(gè)離線存儲(chǔ)來存儲(chǔ)歷史用戶行為，然后每天會(huì)做靜態(tài)特征的計(jì)算，最后放到特征存儲(chǔ)里面給線上的推理服務(wù)用。
與此同時(shí)，系統(tǒng)也會(huì)做離線的樣本拼接，拼接出來的樣本會(huì)存到樣本存儲(chǔ)里面給離線的模型訓(xùn)練使用，離線的模型訓(xùn)練每天會(huì)產(chǎn)生新的模型去驗(yàn)證，然后給到推理服務(wù)使用，這個(gè)模型是一個(gè)T+1的更新。
以上就是一個(gè)經(jīng)典離線推薦系統(tǒng)的架構(gòu)。如果要把它推進(jìn)到實(shí)時(shí)推薦系統(tǒng)里面，主要要做以下三件事情：
- 特征計(jì)算
靜態(tài) T+1 特征計(jì)算到實(shí)時(shí)特征計(jì)算。 - 樣本生成
離線 T+1 樣本生成到實(shí)時(shí)樣本生成。 - 模型訓(xùn)練
離線訓(xùn)練 T+1 更新到增量訓(xùn)練實(shí)時(shí)更新。
（二）阿里巴巴搜推廣在線機(jī)器學(xué)習(xí)流程
阿里巴巴搜推廣已經(jīng)上線了這樣的實(shí)時(shí)推薦系統(tǒng)，它的整個(gè)流程其實(shí)跟離線的推薦系統(tǒng)是類似的，主要區(qū)別是整個(gè)過程都實(shí)時(shí)化了。
?
如上所示，這套系統(tǒng)主要有三方面的特性：
時(shí)效性：大促期間，全流程實(shí)時(shí)更新。
靈活性：根據(jù)需求，隨時(shí)調(diào)整特征和模型。
可靠性：系統(tǒng)穩(wěn)定、高可用，上線效果保證。
用戶可以做到非常有時(shí)效性地更新模型、特征，在大促的期間，可以隨時(shí)調(diào)整特征和模型，表現(xiàn)出來的效果也很好。
（三）實(shí)時(shí)推薦系統(tǒng)架構(gòu)
實(shí)時(shí)推進(jìn)系統(tǒng)的架構(gòu)應(yīng)該長成什么樣子？
如上圖所示，相比于剛才經(jīng)典的離線推薦系統(tǒng)，實(shí)時(shí)推薦架構(gòu)發(fā)生了一些變化。首先，消息隊(duì)列生成的數(shù)據(jù)，除了進(jìn)到離線存儲(chǔ)保存歷史行為以外，系統(tǒng)還會(huì)把這個(gè)消息隊(duì)列里面的消息讀出來兩份，其中一份拿去做實(shí)時(shí)的特征計(jì)算，也是會(huì)放到特征存儲(chǔ)里面，另外一份是會(huì)放到實(shí)時(shí)樣本拼接里面，跟線上的推理服務(wù)使用的用戶特征進(jìn)行一個(gè)雙流Join，這樣能夠得到一個(gè)實(shí)時(shí)的樣本。
在這種情況下，存儲(chǔ)到實(shí)時(shí)系統(tǒng)的樣本可以同時(shí)被拿來做離線的模型訓(xùn)練，也可以拿來做實(shí)時(shí)的模型訓(xùn)練。
不管是離線的還是實(shí)時(shí)的模型訓(xùn)練，它們生成的模型都會(huì)被放到模型存儲(chǔ)里面，并經(jīng)過模型驗(yàn)證最后上線。
離線模型訓(xùn)練是天級(jí)別的，但實(shí)時(shí)模型訓(xùn)練可能是分鐘級(jí)、小時(shí)級(jí)甚至是秒級(jí)的。這個(gè)時(shí)候離線的模型訓(xùn)練會(huì)天級(jí)別產(chǎn)生一個(gè)Base Model給到實(shí)時(shí)的模型訓(xùn)練，然后再去做增量的模型更新。
整個(gè)的架構(gòu)里面有一點(diǎn)需要提到的是，推理服務(wù)在使用這個(gè)特征存儲(chǔ)里面拿過來的特征做推理的同時(shí)，它還需要把本次做推理所用的特征也加上Request ID送到消息隊(duì)列里面。這樣的話實(shí)時(shí)樣本拼接的時(shí)候，當(dāng)產(chǎn)生一個(gè)正樣本，比方說用戶展示了某一個(gè)廣告，然后點(diǎn)擊了之后它是一個(gè)正樣本，這時(shí)候才能夠知道當(dāng)時(shí)用了哪些特征給用戶推薦的廣告，所以這個(gè)特征信息是需要推理服務(wù)保留下來，送到實(shí)時(shí)樣本里面做樣本拼接，才能生成一個(gè)很好的樣本。
這個(gè)架構(gòu)里面可以看到，相比于經(jīng)典的離線推薦系統(tǒng)，在綠色框的部分都是實(shí)時(shí)的部分，有一些部分是新加的，有一些部分是把原來離線的部分變成了實(shí)時(shí)的部分。比如實(shí)時(shí)特征計(jì)算是新加的，實(shí)時(shí)樣本拼接是把原來的離線樣本拼接的部分變成了實(shí)時(shí)，實(shí)時(shí)模型訓(xùn)練是新加的，模型驗(yàn)證也是同樣的道理，是把原來的離線模型驗(yàn)證，變成了實(shí)時(shí)的模型驗(yàn)證。
（四）基于 Flink + Hologres 的實(shí)時(shí)推薦方案
如果要實(shí)現(xiàn)剛才的實(shí)時(shí)推薦系統(tǒng)架構(gòu)，會(huì)用到一些什么樣的系統(tǒng)？
如上圖所示，消息隊(duì)列用的是Kafka，離線的存儲(chǔ)假設(shè)用的是HDFS。不管是實(shí)時(shí)特征計(jì)算還是離線特征計(jì)算，現(xiàn)在都可以用Flink來進(jìn)行計(jì)算，利用Flink流批一體的能力，能夠保證實(shí)時(shí)和離線的特征計(jì)算所產(chǎn)生的結(jié)果是一致的。
Hologres在這里的作用是特征存儲(chǔ)，Hologres特征存儲(chǔ)的好處是可以提供非常高效的點(diǎn)查，另一個(gè)就是在做實(shí)時(shí)特征計(jì)算的時(shí)候，經(jīng)常會(huì)產(chǎn)生一些不準(zhǔn)確的特征，需要在后期對(duì)這些特征進(jìn)行一些修正。可以通過Flink加Hologres的機(jī)制進(jìn)行很好的特征的修正。
同樣的道理，在推理服務(wù)這一側(cè)，通過保留用來做推理的特征，放到后面的樣本拼接里面,這里的消息隊(duì)列也會(huì)使用Kafka。樣本拼接這個(gè)事情會(huì)用Flink來做，Flink一個(gè)非常經(jīng)典的應(yīng)用場景做雙流Join。把樣本給拼接出來后，在把特征給加上，接著把算好的樣本同樣也放進(jìn)Hologres里面做樣本的存儲(chǔ)。
在樣本存儲(chǔ)的情況下，Hologres里面的樣本既可以拿來做實(shí)時(shí)的模型訓(xùn)練，通過讀取Hologres的Binlog來做實(shí)時(shí)的模型訓(xùn)練，也可以通過Hologres批量的Scan去做離線的模型訓(xùn)練。
不管是在線還是離線的模型訓(xùn)練，都可以用Flink或者是FlinkML，也就是Alink來做。如果是傳統(tǒng)機(jī)器學(xué)習(xí)的話，也可以用TensorFlow來做深度學(xué)習(xí)的模型訓(xùn)練，這樣的模型還是可能會(huì)存到HDFS，然后通過Flink和TensorFlow做模型的驗(yàn)證，最后做線上的推理服務(wù)。
線上推理服務(wù)很多用戶會(huì)有自己的推理引擎，如果有可以用，如果想用Flink和TensorFlow的話也可以直接使用。
（五）實(shí)時(shí)特征計(jì)算及推理 （Flink + Hologres）
首先我們來看實(shí)時(shí)特征計(jì)算和推理的過程，如上圖所示。
剛才提到我們會(huì)把實(shí)時(shí)的用戶行為采集下來，送到Flink里面去做實(shí)時(shí)特征計(jì)算，然后存進(jìn)Hologres里面給線上推理服務(wù)使用。
這里的實(shí)時(shí)特征可能包含：
- 用戶最近 5 分鐘的瀏覽記錄
1）商品、文章、視頻
2）停留時(shí)長
3）收藏、加購、咨詢，評(píng)論 - 最近 10 分鐘每個(gè)品類中點(diǎn)擊率最高的 50 個(gè)商品
- 最近 30 分鐘瀏覽量最高的文章、視頻、商品
- 最近 30 分鐘搜索量最高的 100 個(gè)詞
對(duì)于搜推廣業(yè)務(wù)，都可以用這樣的實(shí)時(shí)特征來更好的獲得推薦效果。
（六）實(shí)時(shí)樣本拼接（Flink + Hologres）
再往下我們會(huì)看實(shí)時(shí)樣本拼接的部分，如下圖所示。
實(shí)時(shí)用戶行為會(huì)被采集下來，進(jìn)到Flink里面去做樣本的拼接。這里的樣本拼接包含了兩個(gè)部分，第一個(gè)部分是首先要知道這個(gè)樣本是正樣本還是負(fù)樣本，這是通過分析實(shí)時(shí)用戶行為的日志來的，我們會(huì)有展示流、點(diǎn)擊流，如果展示流Join點(diǎn)擊流，然后發(fā)現(xiàn)展示的一個(gè)Item被用戶點(diǎn)擊了，那么這就是正樣本。如果我們展示了某個(gè)Item用戶沒有點(diǎn)擊，那么就是一個(gè)負(fù)樣本，這就是我們判斷正負(fù)樣本的過程。
僅僅有正負(fù)樣本的判斷顯然不夠，因?yàn)樵谧鲇?xùn)練的時(shí)候還需要這個(gè)特征，這些特征是從推理服務(wù)過來的，當(dāng)展示某一個(gè)Item的時(shí)候，推理服務(wù)就使用了某一些特征來判斷用戶是否會(huì)對(duì)這個(gè)東西感興趣。這些特征會(huì)放到Kafka里面留存下來，進(jìn)到Flink里面。做樣本拼接的過程當(dāng)中，會(huì)通過Request ID Join上當(dāng)時(shí)去做推薦的所用到這些特征，然后生成一個(gè)完整的樣本放到Hologres里面。
這里會(huì)利用 Flink 多流 Join 能力進(jìn)行樣本拼接，與此同時(shí)也會(huì)做多流同步、正負(fù)樣本、樣本修正。
（七）實(shí)時(shí)模型訓(xùn)練 / 深度學(xué)習(xí) ( PAI-Alink / Tensorflow)
在樣本生成了以后，下一個(gè)步驟就是實(shí)時(shí)的模型訓(xùn)練或者深度學(xué)習(xí)。
如上圖所示，在這種情況下，剛才說到樣本是存在Hologres里面的，Hologres里面的樣本可以用作兩個(gè)用途，既可以用做在線的模型訓(xùn)練，也可以用做離線的模型訓(xùn)練。
在線的模型訓(xùn)練和離線的模型訓(xùn)練可以分別利用Hologres的Binlog和批量Scan的功能去做。從性能上來講，其實(shí)跟一般的消息隊(duì)列或者文件系統(tǒng)去掃描相差并不大。
這里如果是深度模型的話，可以用TensorFlow來做訓(xùn)練。如果是傳統(tǒng)機(jī)器學(xué)習(xí)模型的話，我們可以用Alink或者說FlinkML來做訓(xùn)練，然后進(jìn)到HDFS存儲(chǔ)，把模型給存儲(chǔ)起來，接著再通過Flink或者TensorFlow來做模型的驗(yàn)證。
上述過程是實(shí)際搭建實(shí)時(shí)模型和深度模型訓(xùn)練可以用到的一些技術(shù)。
（八）Alink–Flink ML（基于Flink的機(jī)器學(xué)習(xí)算法）
這里簡單的介紹一下Alink，Alink是基于Flink的一個(gè)機(jī)器學(xué)習(xí)算法庫，目前已經(jīng)開源，正在向 Apache Flink 社區(qū)進(jìn)行貢獻(xiàn)中。
如上圖所示，Alink (Flink ML)相比于Spark ML來講有兩個(gè)特色：
（九）離線特征回填 (Backfill)
介紹完訓(xùn)練部分，再來看離線特征回填。這個(gè)過程其實(shí)是說在上線實(shí)時(shí)特征以后，需要上線新的特征，應(yīng)該怎么做？
如上圖所示，一般會(huì)分成兩步。第一步會(huì)在實(shí)時(shí)的系統(tǒng)里面先把新的特征給加上，那么從某一個(gè)時(shí)刻開始，Hologres里面存儲(chǔ)生成的特征都是有新的特征了。對(duì)于那些歷史數(shù)據(jù)怎么辦？這個(gè)時(shí)候就需要重新做一個(gè)特征回填，用HDFS里面存的歷史行為數(shù)據(jù)跑一個(gè)批量的任務(wù)，然后把歷史上的一些特征給補(bǔ)上。
所以離線特征回填在這個(gè)架構(gòu)圖里面也是由Flink的離線特征計(jì)算來完成的，從HDFS里面把歷史行為數(shù)據(jù)讀出來，然后去算一些離線的特征，把過去的歷史消息里面的特征給補(bǔ)上。
基于Apache Flink + Hologres的實(shí)時(shí)推薦系統(tǒng)關(guān)鍵技術(shù)
剛才的架構(gòu)里面所用到的關(guān)鍵技術(shù)比較多，接下來主要講兩個(gè)點(diǎn)。
（一）可撤回訂正的特征和樣本
第一個(gè)點(diǎn)是可撤回訂正的特征和樣本，如上圖所示。
圖中有下部陰影的區(qū)域里面，通過Flink和Hologres配合，會(huì)進(jìn)行一些樣本和特征的撤回和訂正。
為什么需要特征和樣本的訂正？
- 實(shí)時(shí)日志存在亂序
例如某個(gè)用戶點(diǎn)擊事件由于系統(tǒng)延遲晚到產(chǎn)生 False Negative 樣本。 - 一般通過離線作業(yè)重新計(jì)算離線樣本
重新跑整個(gè)離線樣本計(jì)算 - 通過 Apache Flink + Hologres 撤回機(jī)制點(diǎn)更新
僅更新需要更正的特征和樣本
實(shí)時(shí)日志有可能會(huì)存在一些亂序，有些流可能到得早一些，有些流可能到得晚一些。在這種情況下，在做多流Join的時(shí)候就有可能會(huì)由于系統(tǒng)的延遲、晚到而產(chǎn)生一些False Negative樣本。
舉個(gè)例子，比如在做展示和點(diǎn)擊流Join的時(shí)候，可能一開始認(rèn)為用戶并沒有點(diǎn)擊某一個(gè)廣告，后來發(fā)現(xiàn)用戶點(diǎn)擊了，但是這條事件到的時(shí)間晚了。在這種情況中，一開始會(huì)告訴下游用戶沒有點(diǎn)擊，這是一個(gè)False Negative，后面發(fā)現(xiàn)用戶其實(shí)點(diǎn)擊了，因此需要對(duì) False Negative做修正。當(dāng)發(fā)生這種情況，需要對(duì)之前的樣本做撤回或者更新，去告訴它之前的樣本不是負(fù)樣本，而是正樣本。
基于上述這種情況，我們需要整套鏈路上面有一個(gè)撤回的能力，需要逐級(jí)告訴下游之前的錯(cuò)誤，需要把它給修正，通過Apache Flink + Hologres配合可以完成這樣一個(gè)機(jī)制。
為什么要做這樣一件事情？
以前產(chǎn)生這種False Negative樣本的時(shí)候，一般都是通過離線作業(yè)重新計(jì)算離線樣本進(jìn)行更正。這種方式的代價(jià)是可能需要重新跑整個(gè)離線的樣本計(jì)算，但最終目的其實(shí)僅僅是修正所有樣本里其中很小的一部分樣本，因此這個(gè)代價(jià)是比較高昂的。
通過Apache Flink + Hologres實(shí)現(xiàn)的機(jī)制，可以做到對(duì)False Negative樣本進(jìn)行點(diǎn)狀的更新，而不是重新跑整個(gè)樣本，這種情況下，更正特征和樣本的代價(jià)就會(huì)小很多。
（二）基于事件的流批混合工作流
在這個(gè)架構(gòu)里另一個(gè)關(guān)鍵技術(shù)是基于事件的流批混合工作流，它是什么意思？
看這個(gè)圖，除了剛才所示那些系統(tǒng)之外，這也是一個(gè)非常復(fù)雜的工作流。因?yàn)椴煌南到y(tǒng)之間，它可能存在依賴關(guān)系和調(diào)度關(guān)系，有的時(shí)候是數(shù)據(jù)依賴，有的時(shí)候是控制依賴。
例如，我們可能會(huì)周期性或者定期去跑一些離線的靜態(tài)特征計(jì)算，有可能是做特征回填，也有可能是更正實(shí)時(shí)特征產(chǎn)生的問題，但可能是默認(rèn)周期性地跑，也有可能是手動(dòng)觸發(fā)地跑。還有的時(shí)候是當(dāng)離線模型訓(xùn)練生成之后，需要去觸發(fā)在線模型驗(yàn)證的動(dòng)作，也有可能是在線的模型訓(xùn)練生成以后要去觸發(fā)在線模型訓(xùn)練的動(dòng)作。
還有可能是樣本拼接到了某一個(gè)點(diǎn)，比如上午10點(diǎn)樣本拼接完成之后，想要告訴模型訓(xùn)練說，上午10點(diǎn)之前的樣本都拼接好了，希望想跑一個(gè)批量離線訓(xùn)練的任務(wù)，把昨天早上10點(diǎn)到今天早上10點(diǎn)的數(shù)據(jù)做離線的模型訓(xùn)練。這里它是由一個(gè)流任務(wù)觸發(fā)一個(gè)批任務(wù)的過程。在剛才提到的批量模型訓(xùn)練生成之后，需要放到線上做模型驗(yàn)證的過程當(dāng)中，它其實(shí)是一個(gè)批任務(wù)觸發(fā)流任務(wù)的過程，也會(huì)線上模型訓(xùn)練產(chǎn)生的模型，需要去線上模型訓(xùn)練進(jìn)行驗(yàn)證，這是流任務(wù)觸發(fā)流任務(wù)的過程。
所以在這個(gè)過程當(dāng)中，會(huì)涉及到很多不同任務(wù)之間的交互，這里叫做一個(gè)比較復(fù)雜的工作流，它既有批的任務(wù)又有流的任務(wù)，所以它是一個(gè)流批混合的工作流。
（三）Flink AI Flow
如何做到流批混合的工作流實(shí)現(xiàn)？
使用的是Flink AI Flow，它是一個(gè)大數(shù)據(jù)加AI頂層工作流抽象。
如上圖所示，一個(gè)工作流通常可以分為Workflow定義和Workflow執(zhí)行這兩個(gè)步驟。
Workflow定義會(huì)定義Node和Relation，即定義節(jié)點(diǎn)和節(jié)點(diǎn)之間的關(guān)系。在Flink AI Flow里面，我們把一個(gè)節(jié)點(diǎn)定義成一個(gè)Logical Processing Unit，然后把這個(gè)節(jié)點(diǎn)之間的關(guān)系定義成Event driven conditions。在這樣的抽象下面，在Workflow執(zhí)行層面做了一個(gè)基于事件的調(diào)度。
抽象嚴(yán)格來，在一個(gè)系統(tǒng)里面會(huì)有很多的事件，把這些事件組合到一起，可能會(huì)滿足某一些條件，當(dāng)滿足一個(gè)條件的時(shí)候，會(huì)產(chǎn)生一些動(dòng)作。
例如，一個(gè)工作流中可能有一個(gè)任務(wù)A，它可能會(huì)監(jiān)聽這個(gè)系統(tǒng)里面各種各樣的事件。當(dāng)事件1發(fā)生，然后發(fā)生了事件2，接著發(fā)生了事件3，當(dāng)事件按照這么一個(gè)序列發(fā)生之后，需要做啟動(dòng)任務(wù)A的動(dòng)作，事件123按序發(fā)生是條件。
通過這樣的抽象，可以很好地把以前傳統(tǒng)工作流和帶有流作業(yè)的工作流整合起來。因?yàn)橐郧皞鹘y(tǒng)的工作流里都是基于作業(yè)狀態(tài)發(fā)生變化進(jìn)行調(diào)度，一般是作業(yè)跑完了，然后去看怎么跑下一個(gè)作業(yè)。這個(gè)方式的問題是如果作業(yè)是一個(gè)流作業(yè)，那么這個(gè)作業(yè)永遠(yuǎn)跑不完，這個(gè)工作流無法正常工作。
在基于事件的調(diào)度里面，很好地解決了這個(gè)問題。將不再依賴作業(yè)的狀態(tài)發(fā)生變化來進(jìn)行工作流調(diào)度，而是基于事件來做。這樣的話即使是一個(gè)流作業(yè)，它也可以產(chǎn)生一些事件，然后告訴調(diào)度器做一些其他的事情。
為了完成整個(gè)調(diào)度語義，還需要一些支持服務(wù)，協(xié)助完成整個(gè)調(diào)度語義的支持服務(wù)包括：
- 元數(shù)據(jù)服務(wù)（Metadata Service）
- 通知服務(wù)（Notification Service）
- 模型中心（Model Center）
下面來分別看一下這些支持服務(wù)的內(nèi)容。
（四）元數(shù)據(jù)服務(wù)/Metadata Service
元數(shù)據(jù)服務(wù)是管理數(shù)據(jù)集，在工作流里面希望用戶不用非常繁瑣地找到自己的數(shù)據(jù)集，可以幫用戶管理數(shù)據(jù)集，用戶要用的時(shí)候給一個(gè)名字就可以。
元數(shù)據(jù)服務(wù)也會(huì)管理項(xiàng)目（Project），這里的Project是指Flink AI Flow里面的Project，一個(gè)Project里面可以含有多個(gè)工作流，管理Project最主要的目的是為了保證工作流能夠被復(fù)現(xiàn)。
在元數(shù)據(jù)服務(wù)里面，還會(huì)管理工作流和作業(yè)，每個(gè)工作流里面可能會(huì)涉及到很多的作業(yè)。除此之外，也會(huì)管理模型血緣，可以知道模型的版本是由哪一個(gè)工作流當(dāng)中的哪一個(gè)作業(yè)生成的，最后也支持用戶定義一些自定義實(shí)體。
（五）通知服務(wù)/Notification Service
第二個(gè)服務(wù)是通知服務(wù)，它是一個(gè)帶主鍵的事件和事件監(jiān)聽。
舉個(gè)例子，如上圖所示。一個(gè)客戶端希望監(jiān)聽一個(gè)事件，這個(gè)事件的Key是模型。如果 Key被更新的時(shí)候，監(jiān)聽的用戶就會(huì)收到一個(gè)call back，會(huì)告訴他有一個(gè)事件被更新了，那個(gè)事件的主鍵是模型，Value是模型的URI，版本號(hào)是1。
這里能夠起到的一個(gè)作用就是如果驗(yàn)證一個(gè)作業(yè)，它可以去監(jiān)聽Notification Service。當(dāng)有一個(gè)新模型生成的時(shí)候，需要被通知然后對(duì)這個(gè)模型進(jìn)行驗(yàn)證，所以通過Notification Service就可以做這樣的事情。
（六）模型中心/Model Center
模型中心做的是模型多版本的管理，參數(shù)的記錄，包括模型指標(biāo)的追蹤和模型生命周期的管理，還有一些模型可視化的工作。
舉個(gè)例子闡述Flink AI Flow是如何把實(shí)時(shí)推薦系統(tǒng)里面復(fù)雜的工作流，用一個(gè)完整的工作流描述出來。
如上所示，假如有一個(gè)DAG，它里面包含了模型的訓(xùn)練，模型的驗(yàn)證以及在線推理這三個(gè)作業(yè)。
首先，通過Scheduler模型訓(xùn)練的作業(yè)，在提交上去之后，Scheduler會(huì)到Metadata Service里面去更新作業(yè)的狀態(tài)，變成一個(gè)待提交的狀態(tài)。假設(shè)環(huán)境是K8S Cluster，那么它會(huì)提交到Kubernetes上去跑這樣一個(gè)訓(xùn)練作業(yè)。
訓(xùn)練作業(yè)跑起來之后，可以通過作業(yè)狀態(tài)監(jiān)聽器去更新作業(yè)的狀態(tài)。假使這個(gè)作業(yè)是一個(gè)流式的訓(xùn)練作業(yè)，跑了一段時(shí)間以后會(huì)生成一個(gè)模型，這個(gè)模型會(huì)注冊(cè)到模型中心。注冊(cè)完了以后，模型中心會(huì)發(fā)出一個(gè)事件，表示有一個(gè)新的模型版本被注冊(cè)了，這個(gè)事件會(huì)到Scheduler, Scheduler會(huì)監(jiān)聽這些事件。
之后Scheduler就會(huì)去看，當(dāng)收到這個(gè)事件的時(shí)候，有沒有一些條件被滿足了，然后需要做一些什么樣的動(dòng)作。有一個(gè)模型生成的時(shí)候，Scheduler需要去對(duì)這個(gè)模型進(jìn)行驗(yàn)證，這個(gè)條件被滿足以后，需要去拉起一個(gè)作業(yè)，這個(gè)作業(yè)就是一個(gè)模型驗(yàn)證的作業(yè)。
模型驗(yàn)證作業(yè)被拉起之后，它會(huì)到模型中心找到最新被生成的一個(gè)模型版本，然后對(duì)它去進(jìn)行模型的驗(yàn)證。假設(shè)模型驗(yàn)證通過了，這個(gè)模型驗(yàn)證是個(gè)批作業(yè)，它會(huì)告訴Model Center模型被Validated了，這個(gè)時(shí)候模型中心就會(huì)發(fā)送一條Model Validated Version Event給Scheduler，模型被更新了以后，Scheduler會(huì)去看Model Validated，觸發(fā)拉起線上的推理服務(wù)。推理服務(wù)拉起之后，它會(huì)到模型中心里面把剛剛被Validated過的模型拉過來做推理。
假設(shè)推理服務(wù)也是一個(gè)流的作業(yè)，也是一直跑在那里。過了一段時(shí)間之后，線上的流的訓(xùn)練作業(yè)又生成了一個(gè)新的模型，剛才那條路又會(huì)再走一遍，它會(huì)有一個(gè)模型生成的一個(gè)New Model Version Validated，它又會(huì)被Scheduler聽到，Scheduler又拉起一個(gè)Validated作業(yè)，Job2又會(huì)被拉起，拉起之后Validated作業(yè)又會(huì)去驗(yàn)證模型，有可能這個(gè)模型驗(yàn)證又通過了，又會(huì)發(fā)送一條模型New Model Version Validated給模型中心，模型中心會(huì)把這個(gè)Event又給到 Scheduler。這個(gè)時(shí)候，Scheduler會(huì)看到推理作業(yè)其實(shí)已經(jīng)起在那里了，可能就什么都不做。
推理作業(yè)同時(shí)也在監(jiān)聽著Model Version Validated事件，當(dāng)它收到這個(gè)事件的時(shí)候，會(huì)去做的一件事情就是到模型中心里面重新加載最新的被Validated過的事件。
通過這個(gè)例子，解釋了為什么需要流批混合的調(diào)度器和工作流，來實(shí)現(xiàn)端到端的實(shí)時(shí)推薦系統(tǒng)架構(gòu)里所有作業(yè)、工作流的串聯(lián)。
原文鏈接
本文為阿里云原創(chuàng)內(nèi)容，未經(jīng)允許不得轉(zhuǎn)載。
總結(jié)
以上是生活随笔為你收集整理的实时数仓入门训练营：基于 Apache Flink + Hologres 的实时推荐系统架构解析的全部內(nèi)容，希望文章能夠幫你解決所遇到的問題。
- 上一篇： 深度| 每秒1.4亿次！再度刷新TPS记
- 下一篇： 如何让技术想法更容易被理解?