導讀：Flink Machine Learning Library 提供了機器學習相關的 API 與基礎設施，友善使用者建構機器學習的工作流。使用者可以在 Flink ML Library 提供的 API 的基礎之上，将機器學習算法封裝為 Flink 的算子，并且建構相關的工作流來運作訓練或推理服務。
本文主要介紹：
- Flink ML Java API 的設計與使用
- Flink ML Python 介紹
- Flink ML 的生态建設與未來發展方向
分享嘉賓｜黃興勃/周雲峰 阿裡巴巴
編輯整理｜晏世千
出品平台｜DataFunTalk
01
Flink ML Java API 的設計與使用
1. 為什麼需要 Flink ML API？
為了友善複用算法子產品，将每個機器學習算法封裝為一個子產品，通過這種方式來減少重複開發進而提升社群合作的效率。同時使用者隻需要調用 API 即可使用相應的算法，而算法的實作細節對使用者來說是隐藏的，同時不同的算法使用同一套 API 進行描述，統一算法的使用體驗，有助于提升業務的開發效率。
另外，可以将不同的算法子產品進行組合封裝，進而可以實作包含多個機器學習子產品的複雜任務，同時除了通過程式設計方式之外，API 還支援以 JSON 格式來表達不同算法子產品的配置群組合，有利于提升算法的開發效率。
2. Flink ML 的 API 設計
WithParams 接口是所有的 API 中，最上層的接口，它是 Flink ML 用來存取參數的 API。通過繼承這一接口及其子類，不同的算子能夠通過遵循同一規則的 API 來設定參數，并且可以将參數通過 JSON 的格式進行存取。Flink ML 将一些常見的算法配置參數，比如訓練時候的疊代數，或者說是訓練時候的收斂的門檻值等，以通用的 WithParams 子類的方式來提供，算法開發者在實作算法的時候，就可以通過繼承這些子類的方式來使實作類具備這些參數所需的功能，進而減少重複開發的工作量。
Flink ML 提供的 Stage 接口，代表的是一個算法子產品的基本節點。Stage 提供了用來從檔案系統中儲存和恢複及 Save Load 相關的方法，這些方法能夠輔助将一個算法子產品的配置參數以及訓練後得到的有界資料流儲存檔案之中，用于備份或者在不同的實體機之間進行傳輸。
Stage 接口之後，根據算法子產品的目的和功能的不同，Flink ML 提供了兩套子接口，其中 Estimator 所代表的算法具備訓練相關的語義，而 AlgoOperator 及其子類代表的是推理相關的語義。Estimator 提供一個 FIT 方法，這個方法可以接受由多個 Table 組成的訓練資料，然後進行訓練得到相關的模型資料，并且建立出使用這些模型資料進行推理服務的 Model 執行個體。
AlgoOperator 代表的是推理語義，它提供一個無模型語義的抽象類，它的推理語義是通過 API 定義的 Transform 方法，該方法可以接受由多個 Table 所表示的待推理資料，并且傳回相應的推理結果。
Transformer 接口與 AlgoOperator 的接口是相同的，也是通過 Transform 方法來提供推理服務。它和 AlgoOperator 接口的不同點主要在于如果一個算法對于每條輸入的推理資料都會輸出一條相應的推理結果的話，适合使用 Transformer 來進行表達，而如果這個算法對于每條輸入的資料可以産生多條結果，或者說是對于多條推理資料做一個聚合的話，那這樣的算子更适合使用 AlgoOperator 來進行表達。
Model 類是 Transformer 語義，在提供推理功能的同時，它的推理功能适合模型資料相關的，它額外提供了 API 用來設定或者擷取模型資料流。使用者可以使用 SetModelData 來設定一個 Model 執行個體的模型資料。如果說某個 Model 執行個體是通過 Estimator.fit() 方法得到的話，那麼這個執行個體的内部則是已經被設定了 Estimator 算子訓練時所得到的模型資料流。
以上類是算法的核心類，每個具體算法都是 Stage 的某一個子類的具體實作。這裡所提到的所有的 API，他們的方法都是使用 Flink Table API 來表示的，這樣的話，一方面是為了能夠支援流批一體的語義，在實作流批一體的算子的時候，能夠保持 Flink ML API 的相對穩定，同時能夠讓 Flink ML 使用 Flink 對于 Table 相關的一些優化方案，并且這裡的每一個 API 都使用多個table來表示參數或者傳回值，這也使 Flink ML API 能夠表示多輸入多輸出算相關的意義。
3. Flink ML 訓練與部署模型進行推理流處理的流程
使用上述核心的 Flink ML API 來訓練或者部署模型進行推理資料流的處理流程如下。
假設使用者的資料可以分為靜态資料和動态資料，其中靜态資料是從檔案系統，例如 HDFS 中所讀取的資料，動态資料是從消息隊列，例如 Kafka 中所讀取的資料，這些資料通常會首先通過由 AlgoOperator 或 Transformer 所代表的特征處理算子進行預處理之後得到一些訓練資料，包含訓練語義的Estimator 子產品在收到這些訓練資料之後建立對應的 Model 算子，并且把從訓練資料得到的模型資料流輸入到這個 Model 算子中。
同樣的靜态資料和動态資料經過特征處理後得到的推理資料輸入到這個 Model 子產品，使用 Model 的 Transform 方法進行轉換之後，就可以得到預測結果并輸出到系統之外。在這幅圖裡面，每一個方塊可以代表一個或者多個算法子產品，主要提供計算功能，而每個箭頭代表一個或者是多個的 Table 所組成的資料流，而由這些資料流所在的各個子產品是可以部署在不同的在生産環境節點之上的。比如對于 Estimator 子產品來說，可以先在一些離線的節點中完成它的訓練過程，而訓練得到的 Model 執行個體的參數、以及靜态的模型資料可以通過 Save 方法來儲存在檔案系統之中，而實時模型資料流 Table，或者 Data Stream 可以通過 GetModelData 從 Model 中擷取，通過 Flink 的 Sink 來寫入到比如消息隊列之中，然後在前端伺服器節點上可以重新建立這個 Model 執行個體，從檔案系統中和消息隊列中讀取資料，重新部署并提供推理服務。這是分開訓練和部署的大緻流程。
4. 使用 Pipeline/Graph API 建構複雜機器學習作業
為了更好地将 Flink ML 的子子產品組裝為更加具備更加進階功能的複雜結構，Flink ML 提供了 Pipeline 和 Graph 這兩套 API。這兩套 API 它們的共同點在于都是可以用來建構複雜邏輯的機器學習作業，将 Flink ML 的子產品進行組合和封裝，并且将封裝的結果對外表現為一個單獨的子產品，進而隐藏内部的實作細節。Pipeline 與 Graph 差別在于如果使用者想要表示的 Flink ML 作業是多個算子一對一的串聯形式的話，那麼 Pipeline 能夠比較簡單地表示這種結構，而 Graph 主要支援通過 DAG 圖來表示的方式來表達更加複雜的結構。
--
02
Flink ML python 子產品介紹
Flink ML Python 其實是對 Java API 進行了一層封裝，而 Python 使用者在使用Python API 時是無感覺的，算法在執行時還是使用 Java 去執行的。如下的 Flink ML Python 的架構：
--
03
Flink ML 的生态建設與未來發展方向
1. Flink ML 的開源生态建設
（1）獨立的代碼庫與文檔網站
Flink ML 有單獨的代碼庫，不在 Flink 的核心代碼庫中，這樣做是為了避免 Flink 核心代碼庫過于複雜，同時友善 Flink ML 的快速疊代，另外，Flink ML 也提供了相關的文檔網站，主要包括上述提到的基礎設施的原了解釋、API 的使用，以及 Flink ML 提供的各個算子和配置參數以及使用示例。
（2）基礎設施建設
對于希望向 Flink ML 貢獻算法的開發者，Flink ML 提供了性能開發工具，例如算子的性能測試工具，這個工具可以在 Flink ML 的代碼庫中 Flink-ML-Benchmark 子產品下找到。目前性能測試工具支援對算子吞吐量測試，未來還會增加對于延遲和算法的準确度等方面的測試，另外 Flink-ML 增加了一個資料算子資料管理的工具，這個工具一方面使用托管記憶體，友善進行細度的單個算子的記憶體管理，同時它使用記憶體和檔案系統的共同存儲，當資料量需要管理的資料量比較小的時候，資料會被完全存儲在記憶體當中，當資料量超過一定的門檻值的時候，多餘的資料被溢出到檔案系統進行存儲，這樣一方面是在減少檔案 IO 的同時，盡量保證當資料量超過記憶體托管上限時不會造成 OutOfMemory 的問題。
Flink-ML 提供了一個資料廣播工具 With Broadcast，它和普通的 Broadcast 差別在于它除了支讀取廣播資料到一個算子的所有 Task 之外，它還支援優先讀取的方式，也就是說當廣播資料被完全讀取之前，它不會去嘗試消費非廣播資料流。這個工具的使用場景包括比如在推理算子之中使用這個工具可以保證推理算子在讀取到完整的初始模型之前不會去嘗試讀取待推理的資料，因為此時它是沒有一個模型資料來對它進行推理的。
（3）Flink-Extended 生态項目組織
在 Flink-PMC 的授權下，Flink-ML 建立了一個中立的生态項目組織Flink-Extend，建立這個項目的目的在于，如果開發者有一些項目希望貢獻給開源到社群，那麼就可以貢獻到這個組織下面，目前已經貢獻了 Deep Learning On Flink 和 Clink 兩個項目。Deep Learning On Flink 可以把 Predict Flow，甚至包括以後 Pytorch 程式打包為 Flink 的 Java 算子，放到 Flink 當中運作。
這樣做的優點是可以把 Flink 的預處理程式和 Tensorflow 深度學習算法相結合，形成端到端的訓練。Clink 項目提供 C++實作的機器學習算子，并且通過 JNI 的方式為 C++ 的算子提供适配 Flink-ML 的 API。這樣做是因為一些使用者希望能夠部署低延遲、高可用的作業來提供推理服務，而在這些領域方面，C++ 比 JAVA 有明顯優勢的，并且通過 JNI 的方式也可以使 C++ 算子提供 JAVA API，這樣也可以減少開發工作量，使一套實作提供兩套接口。
2. Flink ML 的未來發展方向
Flink-ML 未來的發展方向，主要有以下幾個方面。
① Flink-ML 會嘗試将 Alink 算子适配 Flink-ML 的 API，将 Alink 算子遷移到 Flink-ML 的代碼庫當中，友善使用者擷取更多開箱即用的機器學習算子，而在所有能提供的算子之中，會優先豐富特征工程的算法庫。
② 在上述提到的性能測試工具的基礎之上，對 Flink-ML所有的算法的性能，包括吞吐量、延遲和算法精度等方面進行優化，使其達到生産可用。
③ Flink-ML 還會嘗試支援線上算法，在算子中支援輸入流批一體的訓練或者推理資料流。
今天的分享就到這裡，謝謝大家。
｜分享嘉賓｜
黃興勃｜阿裡巴巴 進階開發工程師
Apache Flink Committer
2019 年北京郵電大學碩士畢業，加入阿裡巴巴實時計算團隊，主要從事 PyFlink 及 Flink Python 生态的開發。
周雲峰｜阿裡巴巴 開發工程師
畢業于美國卡耐基梅隆大學，現在阿裡巴巴任開發工程師，目前主要負責Flink ML的開發工作。
｜DataFun新媒體矩陣｜
｜關于DataFun｜
專注于大資料、人工智能技術應用的分享與交流。發起于2017年，在北京、上海、深圳、杭州等城市舉辦超過100+線下和100+線上沙龍、論壇及峰會，已邀請超過2000位專家和學者參與分享。其公衆号 DataFunTalk 累計生産原創文章800+，百萬+閱讀，15萬+精準粉絲。