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