導(dǎo)讀：本次分享的主題為構(gòu)建端到端的聯(lián)邦學(xué)習(xí) Pipeline 生產(chǎn)服務(wù)。聯(lián)邦學(xué)習(xí)的優(yōu)勢(shì)在于能夠保證參與各方在數(shù)據(jù)不出本地，保持?jǐn)?shù)據(jù)獨(dú)立性的情況下，多方共建模型，共同提升機(jī)器學(xué)習(xí)效果。聯(lián)邦機(jī)制下，安全隱私有了優(yōu)勢(shì)，但技術(shù)上也會(huì)面臨更多挑戰(zhàn)。作為一個(gè)工業(yè)級(jí)的框架，端到端的聯(lián)邦學(xué)習(xí) Pipeline 致力于完成高彈性、高性能的聯(lián)邦學(xué)習(xí)任務(wù)，主要包括建模、訓(xùn)練、模型管理、生產(chǎn)發(fā)布和在線推理幾個(gè)方面。本次將為大家分享如何靈活調(diào)度管理復(fù)雜的聯(lián)邦學(xué)習(xí)任務(wù)、可視化聯(lián)邦建模的實(shí)現(xiàn)以及在線聯(lián)邦推理服務(wù)的思考與實(shí)踐，解決實(shí)驗(yàn)性機(jī)器學(xué)習(xí)到實(shí)際生產(chǎn)應(yīng)用落地的難點(diǎn)。

主要分享4個(gè)方面：

• 背景介紹

• 高彈性聯(lián)邦學(xué)習(xí) Pipeline 調(diào)度

• 聯(lián)邦學(xué)習(xí)任務(wù)可視化

• 高性能聯(lián)邦學(xué)習(xí)在線推理服務(wù)

我們?cè)谌粘＝＿^程中，會(huì)遇到哪些需求和痛點(diǎn)？

1. 機(jī)器學(xué)習(xí)任務(wù)編排

我們是如何編排機(jī)器學(xué)習(xí)任務(wù)的：

• 塞滿邏輯的 Python 腳本。把機(jī)器學(xué)習(xí)的一些步驟，如特征工程、預(yù)處理、驗(yàn)證等，寫成一個(gè)符合邏輯的 Python 腳本。
• 步驟要并行、要嵌套。通過多線程、多進(jìn)程等手段來完成。
• N 個(gè)業(yè)務(wù)需要 N 個(gè) Script。生產(chǎn)中我們有非常多的業(yè)務(wù)，每個(gè)業(yè)務(wù)采用的算法也不同，所以會(huì)有多個(gè) Script。
• 外部系統(tǒng)對(duì)接。一般其他平臺(tái)在發(fā)起自動(dòng)化機(jī)器學(xué)習(xí)平臺(tái)任務(wù)時(shí)，系統(tǒng)對(duì)接上是非常困難的，畢竟還沒有那么智能，那么如何更好的寫好 Python 腳本呢？

2. 機(jī)器學(xué)習(xí)任務(wù)狀況觀察

當(dāng)我們千辛萬苦的把任務(wù)運(yùn)行起來，還需要不斷觀察任務(wù)的運(yùn)行情況，運(yùn)行到哪一步？每一步的數(shù)據(jù)輸出是什么樣的？指標(biāo)輸出是什么樣的？Loss、AUC 等指標(biāo)的趨勢(shì)？然后，任務(wù)跑完了嗎？

我們?cè)撊绾斡^察到各種趨勢(shì)，來盡可能做參數(shù)的調(diào)整等操作。這里要大家可以想想，平時(shí)是如何觀察機(jī)器學(xué)習(xí)任務(wù)狀態(tài)的呢？

如果是我，可能就會(huì)看日志。那有沒有需求是看日志不能解決的？如果有，那就加幾行日志！^_^

3. 聯(lián)邦學(xué)習(xí)任務(wù)協(xié)同建模

剛剛介紹的都是一般機(jī)器學(xué)習(xí)任務(wù)中會(huì)遇到的挑戰(zhàn)，在聯(lián)邦機(jī)器學(xué)習(xí)下，因?yàn)樯婕岸喾絽f(xié)同，會(huì)遇到更多的挑戰(zhàn)：

• 啟動(dòng)多方任務(wù)。如需要跟多方協(xié)調(diào)啟動(dòng)任務(wù)，時(shí)間難同步。

• 各方的運(yùn)行狀況。需要通過通信工具，同步任務(wù)進(jìn)度。

• 多方日志、排查問題。如日志是散落的，各個(gè)操作系統(tǒng)遇到的問題也不一樣。

• 多方任務(wù)管理?？赡芊锹?lián)邦學(xué)習(xí)的建模只需要1個(gè)人，而聯(lián)邦機(jī)制下需要3個(gè)人或者更多。

接下來分享下 FATE 是如何嘗試解決這樣的問題，以及當(dāng)前的方案和 Pipeline 調(diào)度。

1. 端到端的聯(lián)邦學(xué)習(xí) Pipeline

這是一個(gè)比較典型的縱向聯(lián)邦學(xué)習(xí) Pipeline 的例子：

交集 ( intersect ) -> 聯(lián)邦統(tǒng)計(jì) -> 聯(lián)邦特征處理 -> Training -> 驗(yàn)證 -> 模型保存 -> 模型發(fā)布 ( 發(fā)布到線上服務(wù) FATE-serving，后面會(huì)介紹 )

2. FATE-Flow：端到端的聯(lián)邦學(xué)習(xí) Pipeline 調(diào)度平臺(tái)

① FATE-Flow 功能

FATE-Flow 是我們自研的聯(lián)邦學(xué)習(xí)調(diào)度平臺(tái)，主要有5項(xiàng)功能：

• 用 DAG 定義聯(lián)邦學(xué)習(xí) Pipeline。DAG 具有比較靈活、彈性的特點(diǎn)，是業(yè)界工作流調(diào)度系統(tǒng)常用的方式；在聯(lián)邦學(xué)習(xí)場(chǎng)景下，DAG 是有難點(diǎn)的，由于協(xié)作的機(jī)制，我們的 Pipeline 是非對(duì)稱的；我們?cè)谠O(shè)計(jì) DSL 時(shí)，考慮更多系統(tǒng)化對(duì)接的情況，所以我們采用了 json 格式的 DAG DSL，然后再采用 DSL-Parser 進(jìn)行解析。

• 聯(lián)邦任務(wù)生命周期管理。對(duì)應(yīng)剛剛所說的痛點(diǎn)，對(duì)聯(lián)邦任務(wù)生命周期進(jìn)行管理，如多方啟停、狀態(tài)檢測(cè)。

• 聯(lián)邦任務(wù)協(xié)同調(diào)度。包括：多方任務(wù)隊(duì)列、分發(fā)任務(wù)、狀態(tài)同步等協(xié)同調(diào)度。

• 聯(lián)邦任務(wù)輸入輸出實(shí)時(shí)追蹤。這個(gè)功能比較普遍，一般的調(diào)度平臺(tái)都會(huì)有，包括：數(shù)據(jù)、模型、自定義指標(biāo)、日志等的實(shí)時(shí)記錄存儲(chǔ)。

• 聯(lián)邦模型管理。聯(lián)邦學(xué)習(xí)模型的管理存在一個(gè)很大的問題，尤其在縱向聯(lián)邦學(xué)習(xí)場(chǎng)景下，就是如何保證多方模型的一致性。

② FATE-Flow 架構(gòu)

FATE-Flow 架構(gòu)：

• DSL Parser：是調(diào)度的核心，通過 DSL parser 可以拿到上下游關(guān)系、依賴等。

• Job Scheduler：是 DAG 層面的調(diào)度，把 DAG 作為一個(gè) Job，把 DAG 里面的節(jié)點(diǎn) run 起來，就稱為一個(gè) task。

• Federated Task Scheduler：最小調(diào)度粒度就是 task，需要調(diào)度多方運(yùn)行同一個(gè)組件但參數(shù)算法不同的 task，結(jié)束后，繼續(xù)調(diào)度下一個(gè)組件，這里就會(huì)涉及到協(xié)同調(diào)度的問題。

• Job Controller：聯(lián)邦任務(wù)控制器

• Executor：聯(lián)邦任務(wù)執(zhí)行節(jié)點(diǎn)，支持不同的 Operator 容器，現(xiàn)在支持 Python 和 Script 的 Operator。Executor，在我們目前的應(yīng)用中拉起 FederatedML 定義的一些組件，如 data io 數(shù)據(jù)輸入輸出，特征選擇等模塊，每次調(diào)起一個(gè)組件去 run，然后，這些組件會(huì)調(diào)用基礎(chǔ)架構(gòu)的 API，如 Storage 和 Federation Service ( API 的抽象 ) ，再經(jīng)過 Proxy 就可以和對(duì)端的 FATE-Flow 進(jìn)行協(xié)同調(diào)度。

• Tracking Manager：任務(wù)輸入輸出的實(shí)時(shí)追蹤，包括每個(gè) task 輸出的 data 和 model。

• Model Manager：聯(lián)邦模型管理器

3. DAG 定義聯(lián)邦學(xué)習(xí) Pipeline

左邊為我們的 DSL，它的結(jié)構(gòu)比較簡(jiǎn)單，我們可以定義一串 Component，通過 Parser 解析出 DAG 圖（如右圖，可以清晰地看到整個(gè)算法流程的架構(gòu)）。

構(gòu)建 DSL 只需要三步：

① Module：模型組件，F(xiàn)ATE 當(dāng)前支持11個(gè)模型組件，基本滿足當(dāng)前 FATE 所支持的所有算法。

② Input：

• data：數(shù)據(jù)輸入

• model：模型輸入

• isometric_model：異構(gòu)模型，當(dāng)前只用于 Feature Selection

③ Output：

• data：輸出輸出

• model：模型輸出

可參考下面的例子：

構(gòu)建 DSL 示例

DSL 怎么工作的呢？它是一個(gè)非?？岬哪K，就像人類的大腦，它是 FATE-Flow 的中心：

• 組件初始化：

① 根據(jù) DSL 定義和任務(wù)配置，解析每個(gè) Component 運(yùn)行參數(shù)

② 分析 DSL 定義 data、model 輸入輸出，提取依賴關(guān)系

• DAG 圖：

① 構(gòu)建依賴關(guān)系鄰接表

② 拓?fù)渑判蜻M(jìn)行 DAG 依賴檢測(cè)，因?yàn)橛脩舳x的 DSL 不一定是有效的

• 調(diào)度協(xié)作：

① 實(shí)時(shí)輸出 Component 無依賴上下游

② Component 依賴度自動(dòng)遞減

• 預(yù)測(cè) DSL 推導(dǎo)：

剔除預(yù)測(cè)階段無用 Component 數(shù)據(jù)，模型依賴傳遞，推導(dǎo)預(yù)測(cè) DSL

4. 聯(lián)邦學(xué)習(xí)任務(wù)多方協(xié)同調(diào)度

聯(lián)邦學(xué)習(xí)任務(wù)多方協(xié)同調(diào)度的流程：

首先，是以任務(wù)提交的一種方式，提交任務(wù)到 Queue，然后 JobScheduler 會(huì)把這個(gè)任務(wù)拿出來給到 Federated TaskScheduler 調(diào)度，F(xiàn)ederated TaskScheduler 通過 Parser 取得下游 N 個(gè)無依賴的 Component，再調(diào)度 Executor ( 由兩部分組成：Tracking Manager 和 Task ) 執(zhí)行，同時(shí)這個(gè)任務(wù)會(huì)分發(fā)到聯(lián)邦學(xué)習(xí)的各個(gè)參與方 Host。聯(lián)邦參與方取得任務(wù)，如果是 New Job，則放入隊(duì)列（參與方會(huì)定期調(diào)度隊(duì)列中的 Job），否則啟動(dòng)多個(gè) Executor 執(zhí)行，Executor 在 run 的過程中，會(huì)利用 Federation API 進(jìn)行聯(lián)邦學(xué)習(xí)中的參數(shù)交互，對(duì)一個(gè)聯(lián)邦學(xué)習(xí)任務(wù)，每一方的 Job id 是保持一致的，每跑一個(gè) Component，它的 Task id 也是一致的。每個(gè) Task 跑完 Initiator TaskScheduler 會(huì)收集各方的狀態(tài)，進(jìn)行下一步的調(diào)度。對(duì)于下一步的調(diào)度策略我們支持：all_succss，all_done，one_succss 等策略。由于我們基于 Task 為最小的調(diào)度單位，所以很容易實(shí)現(xiàn) rerun，specified_task_run 等特定運(yùn)行。

5. 聯(lián)邦任務(wù)多方生命周期管理

分以下幾個(gè)部分：

• Task stat：Task 狀態(tài)信息，如啟動(dòng)時(shí)間、運(yùn)行狀態(tài)、結(jié)束時(shí)間、超時(shí)時(shí)間等

• Task run process：Task 運(yùn)行進(jìn)程

• Life cron checker：Task 生命周期定時(shí)檢測(cè)

• Job controller：聯(lián)邦任務(wù)控制器

• Shutdown：kill process、清理任務(wù)以及同步指令到所有聯(lián)邦參與方，保證聯(lián)邦任務(wù)狀態(tài)一致性

啟動(dòng) Shutdown 的條件：

• 若 Task 運(yùn)行時(shí)間超過配置超時(shí)時(shí)間或默認(rèn)超時(shí)時(shí)間(一般較長(zhǎng))，啟動(dòng) Shutdown

• 若 Task 運(yùn)行進(jìn)程異常終止，啟動(dòng) Shutdown

• 若 Task 正常運(yùn)行終止，啟動(dòng) Shutdown

6. 聯(lián)邦任務(wù)輸入輸出實(shí)時(shí)追蹤

聯(lián)邦任務(wù)輸入輸出實(shí)時(shí)追蹤，首先會(huì)有幾個(gè) Definition 定義：

• metric type：指標(biāo)類型，如 auc，loss，ks 等等

• metric namespace：自定義指標(biāo)命名空間，如 train，predict

• metric name：自定義指標(biāo)名稱，如 auc0，hetero_lr_auc0

• metric data：key-value 形式的指標(biāo)數(shù)據(jù)

• metric meta: key-value 形式的指標(biāo)元信息，支持靈活畫圖

目前的 API 只有4~5個(gè)：

• log_metric_data(metric_namespace, metric_name, metrics)

• set_metric_meta(metric_namespace, metric_name, metric_meta)

• get_metric_data(metric_namespace, metric_name)

• get_metric_meta(metric_namespace, metric_name)

可能以前收集指標(biāo)需要經(jīng)過收集日志等一系列操作，任務(wù)像一座座大山一樣擺在面前，現(xiàn)在則有可能成為我們的搖錢樹，因?yàn)槲覀兛梢钥焖俚氖占鞣N指標(biāo)，提交給需求方。

7. 聯(lián)邦模型管理

左圖中的兩桶“大餅”，分別代表了某一方的模型，每一個(gè)“大餅”則代表了每個(gè)組件的 model，如：Dataio、FeatureBinning、FeatureSelection、FeatureTransform、HeteroLR、Pipeline。這里需要做個(gè) Model Binding 模型的綁定，F(xiàn)ATE-Flow 的做法還是比較簡(jiǎn)單的，我們會(huì)給每套模型賦予一對(duì)標(biāo)志符 model_id 和 model_version 來唯一標(biāo)識(shí)模型，model_id 由用戶自定義的 role 和 party_id 及 model_key 拼接而成，model_version 也是可以自定義的，如果不自定義的話，會(huì)默認(rèn)為 job_id。我們會(huì)有一個(gè)命名為 Pipeline 的模型存儲(chǔ) Pipeline 建模 DSL 及在線推理 DSL。

下面是某個(gè)算法模型數(shù)據(jù)結(jié)構(gòu)的示例：

同時(shí)每個(gè)“大餅”算法模型，也由兩部分組成：ModelParam 和 ModelMeta，也就是參數(shù)和元的信息。

8. 聯(lián)邦模型版本管理

模型版本管理我們參考了 Git 的實(shí)現(xiàn)思路，但是我們沒有做的那么復(fù)雜，是基于多叉樹版本的記錄：

• 支持 commit message；

• 支持分支功能，如 experiment，product，release；

• 支持 tag，如 release；

• 支持 history 查看；

• 支持版本回溯，指定某一版本回滾。

9. 使用樣例

上圖為，F(xiàn)ATE-Flow 的簡(jiǎn)單使用樣例，主要就是使用 FATE-Flow CLI 提交一個(gè) Job，需要提供 Job 的 DSL 描述以及配置文件，那么 FATE-Flow Server 會(huì)返回該 Job 的一些必要信息，尤其唯一 Job Id 比較重要。后面則是查詢 Job 狀態(tài)以及停止 Job 的操作指令，CLI 還支持許多豐富的指令，可以參考 github 上的文檔。

第三部分介紹下聯(lián)邦學(xué)習(xí)建模可視化：

1. FATE-Board

大體的架構(gòu)如右圖，有一些 Job DashBoard 和可視化，兩個(gè)基本的管理和上面的 Web UI。

FATE-Board 作為 FATE 聯(lián)邦建模的可視化工具，旨在跟蹤和記錄聯(lián)邦建模全過程的信息，并通過可視化的方式呈現(xiàn)模型訓(xùn)練過程的變化以及模型訓(xùn)練結(jié)果，幫助用戶簡(jiǎn)單而高效地深入探索模型與理解模型。

2. 建模交互及可視化

其基本步驟為：

① 用戶配置 pipeline，建立 graph、定義 parameters 等；

② 用戶提交 job，返回 job URL，同時(shí)啟動(dòng) job 運(yùn)行，進(jìn)入 web 端查看 fateboard；

③ 觀察 job 運(yùn)行狀態(tài)，查看運(yùn)行時(shí)的統(tǒng)計(jì)信息，包括運(yùn)行進(jìn)度、日志、模型過程輸出等；

④ 查看 job 運(yùn)行完成的結(jié)果，包括模型輸出、模型評(píng)分、日志等內(nèi)容及可視化結(jié)果。

第四部主要講怎么發(fā)揮模型的最大價(jià)值，我們構(gòu)建了高性能的聯(lián)邦學(xué)習(xí)在線推理服務(wù)：

1. FATE-Serving

FATE-Serving 設(shè)計(jì)原則：

• 高性能，基于 GRPC 協(xié)議，批量聯(lián)邦請(qǐng)求，聯(lián)邦參與方模型結(jié)果多級(jí)緩存

• 高可用，無狀態(tài)設(shè)計(jì)，異常降級(jí)功能

• 高彈性，模型 & 數(shù)據(jù)處理 App 動(dòng)態(tài)加載

FATE-Serving 模塊：

• Online FederatedML：高性能在線聯(lián)邦學(xué)習(xí)算法包，在線和離線時(shí)性能是不一樣的，在線的時(shí)候，我們沒有采用 Python，當(dāng)前版本是采用 Java 寫的一套聯(lián)邦學(xué)習(xí)算法。

• Online Federated Pipeline：在線聯(lián)邦推理 Pipeline，比較簡(jiǎn)潔高效，后面會(huì)詳細(xì)介紹。

• Dynamic Loaders：推理節(jié)點(diǎn)動(dòng)態(tài)加載器。目前支持模型和 APP 兩種推理節(jié)點(diǎn)，節(jié)點(diǎn)通過訓(xùn)練或者編譯等方式創(chuàng)建后是序列化后存儲(chǔ)在分布式存儲(chǔ)上的。對(duì)于在線服務(wù)，每個(gè)請(qǐng)求都需要經(jīng)過每個(gè)節(jié)點(diǎn)，為了保證執(zhí)行效率，需要把這些節(jié)點(diǎn)提前從存儲(chǔ)中取出緩存在 Server 的內(nèi)存空間，這也是業(yè)界在線服務(wù)的常用做法。

• Model Manager：在線模型管理器

• Processing-App Factory：數(shù)據(jù)處理節(jié)點(diǎn)工廠。在解決實(shí)際業(yè)務(wù)問題時(shí)，還需在模型預(yù)測(cè)的基礎(chǔ)上在 Pipeline 中引入一些人工或者規(guī)則，打包成一個(gè) App 發(fā)布到 Serving，在預(yù)測(cè)的時(shí)候加入一個(gè)前處理 Processing 和一個(gè)后處理 Processing。

• Multi-Level Cache：多級(jí)緩存管理器

• Snapshot Manager：快照管理，定期將在線模型、Processing-App 信息落庫。

2. 在線聯(lián)邦模型管理

在線聯(lián)邦模型管理，我們提供了兩個(gè) API：pulish load req 和 pulish online AB Test。為什么是兩個(gè) API？因?yàn)椋琍ulish load req 過程只是把整個(gè)模型通過 dynamic loaders 得到一個(gè) model object 放入 model pool 中，是一個(gè)內(nèi)存的池子。聯(lián)邦各方的 model id 和 model version 都是保持一致的（在發(fā)布模型的時(shí)候，只需要在 Guest 側(cè)發(fā)起就可以了，會(huì)自動(dòng)走 proxy，把命令推到其他的 Host 中去），當(dāng)各方都加載完之后，就可以應(yīng)用了。

Pulish online AB Test 主要是做 default binding，比如有些業(yè)務(wù)方并不知道我們的模型版本，需要有一個(gè)默認(rèn)的機(jī)制。

圖中右下角，為動(dòng)態(tài)加載器的一個(gè)基本功能。

3. 在線聯(lián)邦推理 Pipeline

在線聯(lián)邦推理 Pipeline 流程：

Req -> Service -> model selection ( 有三種不同策略：規(guī)定的方式和兩種默認(rèn)的方式 ) -> loader -> processing ( 前處理 ) -> inference Pipeline ( 是縱向model，會(huì)經(jīng)過不同的組件，這里舉例是三個(gè)，然后會(huì)把 user_id （用來做對(duì)齊的）發(fā)到不同的參與方，參與方會(huì)拿自己的半模型，進(jìn)行最后的預(yù)測(cè)，最后把結(jié)果發(fā)到 Federated Network ) -> Processing ( 后處理 )

4. 在線推理服務(wù)緩存

在線推理服務(wù)緩存分為三種：

• Inference result in-process cache：cache key 為 caseid，緩存 120s。

• Remote federated result in-process cache：cache key 為 userid，緩存 600s。

• Remote federated result distributed cache：cache key 為 userid，緩存 24h，跟 A 方的特征更新周期有很大的關(guān)系。

在業(yè)務(wù)的運(yùn)行當(dāng)中，按照這幾個(gè)配比，一天下來發(fā)現(xiàn)命中遠(yuǎn)端聯(lián)邦推理結(jié)果 Cache 的請(qǐng)求量占了28%，也就是說有28%的聯(lián)邦請(qǐng)求都不需要這部分，就會(huì)大大的增加了用戶的體驗(yàn)。

我們還有一個(gè)緩存喚醒的熱身過程，我們會(huì)跑一些離線任務(wù)，然后對(duì)活躍度進(jìn)行統(tǒng)計(jì)，把要緩存的用戶按比例提取出來，在凌晨的時(shí)候發(fā)一些批次請(qǐng)求，讓整個(gè)緩存熱身。

5. 聯(lián)邦模型應(yīng)用生產(chǎn)服務(wù)流程

一個(gè)聯(lián)邦模型應(yīng)用到生產(chǎn)服務(wù)的大致流程：

① 全量加載聯(lián)邦模型，通過 pulish load req 全量加載模型，每方的 Serving 都加載這個(gè)模型。

② 灰度上線聯(lián)邦模型，pulish online AB test 可以指定多少用戶上線 online 模型。（這時(shí)就會(huì)體現(xiàn)出為什么要做 pulish load req，因?yàn)樵?Guest 方灰度上線時(shí)，這個(gè) Serving 請(qǐng)求的 user_id 落到其他參與方，需要找到對(duì)應(yīng)的模型，但是通常情況下 Serving 都是做負(fù)載均衡的，如果灰度這部分的 Serving 已經(jīng)上線了新的模型，其它 Host 方?jīng)]有預(yù)先加載這個(gè)模型，就會(huì)找不到這個(gè)模型，也就是說模型是被所有 Serving 所 load 的，但是并不是所有 Serving 當(dāng)前生效的都是這個(gè)模型。）

③ 聯(lián)邦模型效果驗(yàn)證

④ 全量上線聯(lián)邦模型

最后分享下我們聯(lián)邦學(xué)習(xí)的官網(wǎng)：

https://www./cn/

以及 FATE 的 Github 地址：

https://github.com/FederatedAI/FATE

曾紀(jì)策