摘要：人工智能應用場景中，Flink 在包括特征工程，線上學習，線上預測等方面都有一些獨特優勢，為了更好的支援人工智能的使用場景，Flink 社群以及各個生态都在努力。本文将介紹近期 Flink 在人工智能生态系統中的工作進展，主要内容包括：
- Flink 建構 AI 系統的背景
- Flink ML Pipeline 和算法庫 Alink
- 分析和 AI 的統一工作流（AI Flow）
- Flink 在流運作模式下疊代的架構設計
一. Flink 建構 AI 系統的背景
1. Lambda 架構及批流統一資料處理
首先向大家介紹 Flink 建構 AI 系統的背景。
Lambda 架構是大資料處理領域中一種典型的架構。如上圖所示，圖中上方藍色線代表的是離線批處理的資料流，下方黑色線代表的是實時流模式的資料流。為了兼顧整個大資料進行中的吞吐和實時性，我們通常會将離線的資料處理和實時的資料處理結果進行合并後對外提供服務。
在 Lambda 架構裡，在離線資料處理和實時資料處理過程中，我們會用到兩種不同的計算引擎，同時維護兩份代碼。但這兩個流程對資料的處理邏輯其實是一樣的，這樣會導緻兩個問題：
- 維護代價高 ( 包括代碼的維護代價，及不同的計算引擎對部署的維護代價 )
- 很難保證線上離線處理邏輯一緻
為了解決這兩個問題，實作批和流的資料統一，就有了 Flink 和 Spark。Flink 和 Spark 都可以處理批資料，對流資料的處理可以由 Flink 和 Spark streaming 來完成。這樣就避免了上述維護兩套代碼和兩套系統及邏輯一緻性的問題。
2. 機器學習線上離線處理
在機器學習場景下，線上離線處理也會面臨一些問題。上圖是一個典型的機器學習處理過程。首先會将離線的資料進行預處理和特征工程（如紅框标注所示），然後進行離線的模型訓練，訓練好的模型會推到線上做推理。推理子產品加載模型後，線上的資料也會有進行預處理和特征工程的過程，将處理之後的資料喂給模型做線上推理。
這樣的過程也會面臨如上述大資料進行中的問題，我們同樣會維護離線的資料處理和線上推理的資料處理兩份代碼。
在機器學習領域除了離線的模型訓練以外，還有線上的模型訓練。如下圖所示，我們通常會将預處理的資料寫到一個 Message Queue 中（如 Kafka），然後進行 Online training，training 的過程是持續不斷的，期間會不斷的産生動态的模型，然後推送給線上的推理子產品進行推理。線上的機器學習的特點就是模型的動态更新、持續訓練和不斷驗證。同時需要比較複雜的模型監控，模型部署和模型復原等政策。
于是就産生了機器學習中的 Lambda 架構：
一般來說，線上的模型訓練并不是從頭訓練一個模型，而是通過離線訓練出一個基準的模型，然後推給線上，線上流程再在這個基準模型上進行線上的訓練。這樣同樣存在離線和線上兩份代碼，涉及兩套不同的系統。也會增加維護的複雜度。
那麼 Flink 在這套架構中有什麼樣的價值。Flink 是天生支援批流一體的計算引擎，在線上機器學習領域的預處理過程中，我們很自然地會使用 Flink 進行資料處理，在離線的機器學習訓練中也可以使用 Flink 進行批次的預處理。同時，在線上推理的過程中，也會使用 Flink 進行推理（因為推理是一個對時間比較敏感的過程）。是以 Flink 在機器學習 Lambda 架構中的價值展現在線上的資料的預處理，離線資料的預處理，線上的推理。
是以是否能将機器學習中的 Lambda 架構進行批流統一？
如上圖所示，除了資料處理和推理可以使用 Flink 外，Online training 和 offline training 的過程也可以用 Flink 計算引擎替代。這樣做的好處是：使用者隻需要寫一份代碼就可以同時實作線上和離線的訓練，并且避免了邏輯的不一緻性和維護的難度。線上和離線訓練中我們通常會使用比如一些深度學習的計算架構比如 TensorFlow 或者 Pytorch 運作在 Flink 上進行模型訓練。第二個好處是，我們可以通過使用 Flink 形成一站式的資料處了解決方案。如下圖：
在整個資料上，Flink 提供了豐富的接口，包括 SQL、DataStream、CEP，如果再加上 ML 的接口就可以共享整個資料集，不會涉及到不同的系統之間資料拷貝的過程。此外，在将資料打通之後，我們可以使用 SQL，DataStream 這些豐富的 API 來處理資料。
二. Flink ML Pipeline 和算法庫 Alink
1. Flink AI 生态系統架構
上圖是整個機器學習的 Lambda 架構圖。對應機器學習任務中的不同階段：
- 首先是資料的管理和擷取階段（Data Acquisition），在這個階段 Flink 提供了非常豐富的 connector（包括對 HDFS，Kafka 等多種存儲的支援），Flink 目前還沒有提供對整個資料集的管理。
- 下一個階段是整個資料的預處理（Preprocessing）及特征工程部分，在這個階段 Flink 已經是一個批流統一的計算引擎，并且提供了較強的 SQL 支援。
- 之後是模型訓練過程（Model Training），在這個過程中，Flink 提供了 Iterator 的支援，并且有如 Alink，MLlib 這樣豐富的機器學習庫支援，且支援 TensorFlow，Pytorch 這樣的深度學習架構。
- 模型産出之後是模型驗證和管理階段（Model Validation & Serving），這個階段 Flink 目前還沒有涉足。
- 最後是線上推理階段（Inference），這個階段 Flink 還沒有形成一套完整的方案。同時形成了 Flink ML Pipeline，以及目前正在做的 Flink AI Flow。
2. Flink ML Pipeline
上圖是 Flink ML Pipeline 的介紹，該 Pipeline 主要涉及兩個抽象，第一個是 Transformer 抽象，是對資料預處理和線上推理的抽象。第二個抽象是 Estimator 抽象，主要是對整個模型訓練的抽象。兩個抽象最大的差異是 Transformer 是将一份資料轉化為另一份處理後的資料，而 Estimator 是将資料進行訓練轉化為模型。
3. 算法庫 Alink
Alink 主要是重寫了 Flink 中很多機器學習庫。其有兩個重要的特點，一是 Alink 是基于 Flink 的 ML Pipeline。第二是基于 Flink Table API。Flink Table API 天然就是批流統一的。
ML Pipeline 簡單案例：
4. ML Pipeline 價值
Flink ML Pipeline 最大的價值在于為最終使用者統一了模型訓練和推理的 API，使用者隻需要關心 Estimator 的 Transformer 裡面的邏輯即可。此外，Pipeline 将整個訓練過程進行了持久化，確定了訓練和推理之間的邏輯一緻性，解決了之前 Lambda 架構中維護兩份代碼可能會導緻的邏輯不一緻問題。
三. 分析和AI的統一工作流（AI Flow）
1. AI Flow 背景
在整個機器學習任務中，有一部分是和模型訓練相關的，還有一部分是資料分析，特征工程相關的。是以 AI Flow 的目标是将這整個流程串起來，提供一個端到端的解決方案。
2. AI Flow 概述
AI Flow 的 API 中包含了 Example（對資料的 API），Transformer（對預處理流程及推理的 API），Trainer（訓練流程的 API），Model（模型管理的 API）等。在每個子產品中都會産生一些中間 meta 資料，AI Flow 将這些資料存儲在 Meta Store 中。這些 API 隻是定義了機器學習中的一些處理邏輯，AI Flow 中的 Translator 則将這些邏輯轉化為真正可執行的任務。
我們的目标是實作整個機器學習 lambda 架構的批流統一，使用者寫一份機器學習處理邏輯，可以同時作用于離線的學習過程，同時也可以支援線上的學習過程。Translator 的作用就是将使用者通過 AI Flow API 寫好的 code 轉化為可執行的任務。目前 AI Flow 中包含兩類 Translator，第一類是 Batch Translator，第二類是 Stream Translator。
在 Translator 中還有一個抽象是 Components，包含與 AI Flow API 相對應的一些 Components。這些 Components 可以對應地解析 AI Flow API 中使用者定義的邏輯。通過 Translator 将使用者定義好的邏輯處理完之後，通過 Deployer 把生成的任務部署到不同的環境，包括本地環境，K8S，Yarn 等。
3. AI Flow 原理
使用者通過 AI Flow API 寫的代碼在編譯之後會生成一個邏輯的執行計劃，這些邏輯執行計劃由很多節點組成，每個節點都有對應的 Translator 中的 Component 解析，解析過程有可能會将其解析為單個的 job，或者一個 job 的一部分（即多個節點生成一個 job）。這個邏輯執行計劃會傳遞給 Translator，如果是流式任務，Translator 會将其翻譯為流式的一些 job，同樣的批式任務會翻譯成批相關的一些 job。之後會将一組job組成一個 Execution，傳遞給 Deployer 通過配置運作在本地，K8S 或 Yarn 上，最後生成一些可以運作的任務。
4. AI Flow 的特點和 Flink AI Flow
AI Flow 提供了用于部署生産環境資料分析和機器學習流水線的端到端的 API，提供了批流統一的資料分析和機器學習工作流 API。具體來說有以下幾個特點：
AI Flow：
- 批流統一
- 引擎與平台無關
- 定義執行元件關系
- 定義資料集和 IO 格式規範
Flink AI Flow：
- Flink 作為預設分析引擎
- TensorFlow / Pytorch 作為機器學習引擎
5. 圖檔分類簡單案例
首先通過 Example 對象定義處理前的圖檔資料，然後通過 Transformer 對象定義預處理過程和模型的預測過程，最後将兩個 Transformer 合起來組成一個 Execution 邏輯。
6. AI Flow 總結
AI Flow 提供了部署生産環境資料分析和機器學習流水線的端到端 API，同時 AI Flow 還提供了批流一體的資料分析和機器學習工作流 API。
四. Flink 在流運作模式下疊代的架構設計
1. 背景與動機
Flink 在機器學習中有一個整體的端到端的架構，目前 Flink 是通過 DataStream 和 DataSet 兩套接口分别提供了流處理和批處理的能力。通過前面的講述，我們可以看到 Flink 流處理和批處理的能力可以用于機器學習的資料預處理階段。
其實除了這些通用的流和批的處理之外，Flink 對于機器學習中的模型訓練和圖計算這些複雜的分析場景也提供了原生支援。這些場景的特點是對資料的疊代計算要求較高，目前 Flink 對疊代計算的支援主要是在 DataSet 接口部分，因為 DataSet 整體上是一個批處理的接口，是以 Flink 對疊代分析的支援主要是有限資料疊代分析。
例：基于 DataSet 疊代的 K-Means
以常見的 K-Means 為例，K-Means 算法的兩個輸入為待聚類的點和初始類中心，在 DataSet 疊代時，它會在整個計算圖中添加一個特殊節點來維護待求中心點目前的結果，在每輪疊代中，對待求中心點進行一次更新（将每個點配置設定到較近的中心點，之後重新計算中心點，将計算結果通過一個回邊發送到維護的待求中心點算子，進而支援超過 DAG 計算能力的處理形式），重複多輪疊代直到收斂，輸出最終的中心點。
為了表示這種疊代的計算圖，Flink 中有幾個通用的概念，首先“疊代變量”是在疊代中需要更新的變量，“靜态資料集”是在疊代過程中會多次使用但是不會發生變化的資料，在計算過程中該資料實際上隻會發送一次，Flink 對該資料做了一個基于磁盤的緩存，在每輪疊代時做了一個重放。他們都是普通的 DataSet 對象，代表來一個有限的資料集，在某一個資料集上調用 Iteration 方法，指定疊代終止條件。Flink 會自動将每一輪的疊代邏輯擴充到所有輪，使用者在寫疊代邏輯時不需要考慮收到多輪資料的情況。
為了實作流批一體，Flink 将批處理的能力從 DataSet 的接口遷移到 DataStream 的接口之上，在 DataStream 接口之上，Flink 會引入一個 BoundedStream 的特殊子類，來實作批處理的能力。相應的，如果把批處理的能力遷移過去，DataSet 的疊代的處理能力也要進行遷移，一個選擇是直接平移相應的 DataSet 的實作，但是從前面的介紹可以發現，DataSet 目前的實作有一些問題：
- 首先它不支援多疊代變量或者嵌套疊代的形式，但是在一些算法（如 boosting 算法）中對這兩種疊代也是有需求的。
- 第二點是對于靜态資料，Flink 現在是做了一個基于磁盤的重放，在這種情況下，使用者是有可能基于業務邏輯做一個更高效的緩存的，但是現在在 DataSet 的疊代上無法實作這一點。
- 最後，在 DataSet 上很難實作針對線上算法的模型訓練或者線上流處理的支援。
比如使用每一個小時的資料做一次 K-Means。
利用每小時的資料更新聚類點。
是以我們需要引入一種新的疊代機制，可以相容在有限資料上的疊代，并且支援“無限流上每一部分資料分别進行疊代”的語義。
2. Mini-batch 流式 Iteration
上圖是 Mini-batch 流式 Iteration 示例，對于靜态資料集，其可以視為隻有一個 Mini-batch。對于流式資料，可以将其拆成多個 Mini-batch，他們之間可以獨立并行疊代。算子可以自動将單個 Mini-batch 操作擴充到每一個 Mini-batch 上。
還有一種情況是使用無限的資料集訓練一個統一的模型，這種情況下不同的 Mini-batch 共享同一組狀态（待更新的模型），其執行機制如上圖所示。
在上述疊代過程中，算子需要知道每個 Mini-batch 的疊代終止情況，我們稱之為進度追蹤，其目标是提供各 Mini-batch 各輪疊代終止的通知。這種能力通過 Assigner 節點在資料流中插入特殊的标記消息，然後其它算子對标記消息進行對齊來實作。關于進度追蹤較長的描述，如下：
3. Mini-batch 疊代 API
通過上面的描述可以列出 Mini-batch 疊代 API 的基本架構：
4. 總結
總結來說，我們設計了一種新的基于 Mini-batch 的流式疊代機制，這種疊代機制既可以相容原來的 DataSet 上基于有限資料上的疊代，也可以支援對無限流上每一部分資料分别進行疊代。未來，可以更好地支援線上的機器學習訓練和線上的圖處理的場景。
作者介紹：
陳戊超（仲卓），阿裡巴巴技術專家，加入阿裡巴巴之前曾就職于百度 spider 部門，在阿裡巴巴主要參與搜尋離線業務和機器學習平台建設。目前關注在 Flink 上支援機器學習架構方向。
高赟（雲骞），阿裡巴巴技術專家，2017 年博士畢業于中國科學院後加入阿裡巴巴實時計算 Flink 團隊，主要從事 Flink Runtime 層的設計與研發。
一個預告
3 月 17 日（下周二）晚上 20:00，Flink 社群系列直播課程将邀請 Apache Flink PMC 跟大家分享 PyFlink 的相關進展，您将了解以下内容：
- 細緻分析 PyFlink 所适用的場景，并進行 CDN 日志分析的案例示範
- 首次詳細剖析 PyFlink API 和 UDF 的架構，并分享架構背後的思考
- 首次披露 PyFlink 1.11 功能規劃和性能名額
- 首次談及 PyFlink 的使命願景，一張 PyFlink 大圖勾勒未來願景
- 首次透露 PyFlink 的核心貢獻者的個人聯系資訊，PyFlink 的任何問題可以随時抛來
釘釘掃描下方二維碼即可觀看～