解锁云原生 AI 技能 - 开发你的机器学习工作流
按照上篇文章《解鎖云原生 AI 技能 | 在 Kubernetes 上構建機器學習系統》搭建了一套 Kubeflow Pipelines 之后，我們一起小試牛刀，用一個真實的案例，學習如何開發一套基于 Kubeflow Pipelines 的機器學習工作流。
準備工作
機器學習工作流是一個任務驅動的流程，同時也是數據驅動的流程，這里涉及到數據的導入和準備、模型訓練 Checkpoint 的導出評估、到最終模型的導出。這就需要分布式存儲作為傳輸的媒介，此處使用 NAS 作為分布式存儲。
- 創建分布式存儲，這里以 NAS 為例。此處 NFS_SERVER_IP 需要替換成真實 NAS 服務器地址
開發 Pipeline
由于 Kubeflow Pipelines 提供的例子都是依賴于 Google 的存儲服務，這導致國內的用戶無法真正體驗 Pipelines 的能力。為此，阿里云容器服務團隊提供了基于 NAS 存儲訓練 MNIST 模型的例子，方便您在阿里云上使用和學習 Kubeflow Pipelines。具體步驟分 3 步:
- (1) 下載數據
- (2) 利用 TensorFlow 進行模型訓練
- (3) 模型導出
在這 3 個步驟中，后一個步驟都依賴于前一個步驟而完成。
Kubeflow Pipelines 中可以用 Python 代碼描述這樣一個流程, 完整代碼可以查看 standalone_pipeline.py。
我們在例子中使用了基于開源項目 Arena 的 arena_op ，這是對于 Kubeflow 默認的 container_op 封裝，它能夠實現對于分布式訓練 MPI 和 PS 模式的無縫銜接，另外也支持使用 GPU 和 RDMA 等異構設備和分布式存儲的簡單接入，同時方便從 git 源同步代碼，是一個比較實用的工具 API。
Kubeflow Pipelines 會將上面的代碼轉化成一個有向無環圖 (DAG), 其中的每一個節點就是 Component (組件)，而 Component (組件)之間的連線代表它們之間的依賴關系。從 Pipelines UI 可以看到 DAG 圖:
首先具體理解一下數據準備的部分，這里我們提供了 arena.standalone_job_op 的 Python API, 需要指定該步驟的名稱: name； 需要使用的容器鏡像: image； 要使用的數據以及其對應到容器內部的掛載目錄: data。
這里的 data 是一個數組格式, 如 data=[“user-susan:/training”]，表示可以掛載到多個數據。 其中 user-susan 是之前創建的 Persistent Volume Claim， 而 /training 為容器內部的掛載目錄。
而上述步驟實際上是從指定地址利用 curl 下載數據到分布式存儲對應的目錄 /training/dataset/mnist，請注意這里的 /training 為分布式存儲的根目錄，類似大家熟悉的根 mount 點；而 /training/dataset/mnist 是子目錄。其實后面的步驟可以通過使用同樣的根 mount 點，讀到數據，進行運算。
第二步是利用下載到分布式存儲的數據,并通過 git 指定固定 commit id 下載代碼，并進行模型訓練。
可以看到這個步驟比數據準備要相對復雜一點，除了和第一步驟中的 name， image, data 和 command 一樣需要指定之外，在模型訓練步驟中，還需要指定:
- 獲取代碼的方式: 從可重現實驗的角度來看，對于運行試驗代碼的追本溯源，是非常重要的一環。可以在 API 調用時指定 sync_source 的 git 代碼源，同時通過設定 env 中 GIT_SYNC_REV 指定訓練代碼的 commit id；
- gpu: 默認為 0，就是不使用 GPU；如果為大于 0 的整數值，就代表該步驟需要這個數量的 GPU 數；
- metrics: 同樣是從可重現和可比較的實驗目的出發，用戶可以將需要的一系列指標導出，并且通過 Pipelines UI 進行直觀的顯示和比較。具體使用方法分為兩步：1. 在調用 API 時以數組的形式指定要收集指標的 metrics name 和指標的展示格式 PERCENTAGE 或者是 RAW，比如 metrics=["Train-accuracy:PERCENTAGE"]。 2. 由于 Pipelines 默認會從 stdout 日志中收集指標，你需要在真正運行的模型代碼中輸出 {metrics name}={value} 或者 {metrics name}:{value}, 可以參考具體樣例代碼。
值得注意的是：
在本步驟中指定了和 prepare_data 相同的 data 參數 [“user-susan:/training”]，就可以在訓練代碼中讀到對應的數據，比如 --data_dir /training/dataset/mnist。
另外由于該步驟依賴于 prepare_data，可以在方法中通過指定 prepare_data.output 表示兩個步驟的依賴關系。
最后 export_model 是基于 train 訓練產生的 checkpoint，生成訓練模型：
export_model = arena.standalone_job_op(name="export-model",image="tensorflow/tensorflow:1.11.0-py3",sync_source="https://code.aliyun.com/xiaozhou/tensorflow-sample-code.git",env=["GIT_SYNC_REV=%s" % (commit)],data=data,command="echo %s;python code/tensorflow-sample-code/tfjob/docker/mnist/export_model.py --model_version=%s --checkpoint_path=/training/output/mnist /training/output/models" % (train.output, model_version))export_model 和第二步 train 類似，甚至要更為簡單，它只是從 git 同步模型導出代碼并且利用共享目錄 /training/output/mnist 中的 checkpoint 執行模型導出。
整個工作流程看起來還是很直觀的, 下面就可以定義一個 Python 方法將整個流程貫穿在一起：
@dsl.pipeline 是表示工作流的裝飾器，這個裝飾器中需要定義兩個屬性，分別是 name 和 description。
入口方法 sample_pipeline 中定義了 4 個參數： learning_rate, dropout, model_version 和 commit, 分別可以在上面的 train 和 export_model 階段使用。這里的參數的值實際上是 dsl.PipelineParam 類型，定義成 dsl.PipelineParam 的目的在于可以通過 Kubeflow Pipelines 的原生 UI 將其轉換成輸入表單，表單的關鍵字是參數名稱，而默認值為參數的值。值得注意的是，這里的 dsl.PipelineParam 對應值實際上只能是字符串和數字型；而數組和 map，以及自定義類型都是無法通過轉型進行變換的。
實際上，這些參數都可以在用戶提交工作流時進行覆蓋，以下就是提交工作流對應的 UI：
提交 Pipeline
您可以在自己的 Kubernetes 內將前面開發工作流的 Python DSL 提交到 Kubeflow Pipelines 服務中, 實際提交代碼很簡單:
KFP_SERVICE="ml-pipeline.kubeflow.svc.cluster.local:8888"import kfp.compiler as compilercompiler.Compiler().compile(sample_pipeline, __file__ + '.tar.gz')client = kfp.Client(host=KFP_SERVICE)try:experiment_id = client.get_experiment(experiment_name=EXPERIMENT_NAME).idexcept:experiment_id = client.create_experiment(EXPERIMENT_NAME).idrun = client.run_pipeline(experiment_id, RUN_ID, __file__ + '.tar.gz',params={'learning_rate':learning_rate,'dropout':dropout,'model_version':model_version,'commit':commit})利用 compiler.compile 將 Python 代碼編譯成執行引擎 (Argo) 識別的 DAG 配置文件；
通過 Kubeflow Pipeline 的客戶端創建或者找到已有的實驗，并且提交之前編譯出的 DAG 配置文件。
在集群內準備一個 python3 的環境，并且安裝 Kubeflow Pipelines SDK：
# kubectl create job pipeline-client --namespace kubeflow --image python:3 -- sleep infinity # kubectl exec -it -n kubeflow $(kubectl get po -l job-name=pipeline-client -n kubeflow | grep -v NAME| awk '{print $1}') bash登錄到 Python3 的環境后,執行如下命令，連續提交兩個不同參數的任務：
# pip3 install http://kubeflow.oss-cn-beijing.aliyuncs.com/kfp/0.1.14/kfp.tar.gz --upgrade # pip3 install http://kubeflow.oss-cn-beijing.aliyuncs.com/kfp-arena/kfp-arena-0.4.tar.gz --upgrade # curl -O https://raw.githubusercontent.com/cheyang/pipelines/update_standalone_sample/samples/arena-samples/standalonejob/standalone_pipeline.py # python3 standalone_pipeline.py --learning_rate 0.0001 --dropout 0.8 --model_version 2 # python3 standalone_pipeline.py --learning_rate 0.0005 --dropout 0.8 --model_version 3查看運行結果
登錄到 Kubeflow Pipelines 的 UI: https://{pipeline地址}/pipeline/#/experiments， 比如：
https://11.124.285.171/pipeline/#/experiments
點擊 Compare runs 按鈕，可以比較兩個實驗的輸入、花費的時間和精度等一系列指標。讓實驗可追溯是讓實驗可重現的第一步，而利用 Kubeflow Pipelines 本身的實驗管理能力則是開啟實驗可重現的第一步。
總結
實現一個可以運行的 Kubeflow Pipeline 需要的步驟是：
- 構建運行時代碼：通常是為每個步驟構建容器鏡像，作為 Pipelines 和真正執行業務邏輯代碼之間的適配器。它所做的事情為獲取 Pipelines 上下文的輸入參數，調用業務邏輯代碼，并且將需要傳遞到下個步驟的輸出按照 Pipelines 的規則放到容器內的指定位置，由底層工作流組件負責傳遞。 這樣產生的結果是運行時代碼與業務邏輯代碼會耦合在一起。可以參考 Kubeflow Pipelines 的例子；
- 構建客戶端代碼：這個步驟通常是長成下面的樣子, 熟悉 Kubernetes 的朋友會發現這個步驟實際上就是在編寫 Pod Spec：
利用原生定義的 dsl.container_ops 的好處在于靈活，由于開放了和 Pipelines 的交互接口，用戶可以在 container_ops 這個層面做許多事情。但是它的問題在于：
- 復用度低。每個 Component 都需要構建鏡像和開發運行時代碼；
- 復雜度高。使用者需要了解 Kubernetes 的概念，比如 resource limit, PVC, node selector 等一系列概念；
- 支持分布式訓練困難。由于 container_op 為單容器操作，如果需要支持分布式訓練就需要在 container_ops 中提交和管理類似 TFJob 的任務。這里會帶來復雜度和安全性的雙重挑戰，復雜度比較好理解，安全性是說提交 TFJob 這類任務的權限會需要開放額外的權限給 Pipeline 的開發者。
另一種方式是使用 arena_op 這種可以重用的 Component API，它使用通用運行時代碼，可以免去重復構建運行時代碼的工作；同時利用通用一套的 arena_op API 簡化用戶的使用；也支持 Parameter Server 和 MPI 等場景。建議您使用這種方式編譯 Pipelines。
總結
以上是生活随笔為你收集整理的解锁云原生 AI 技能 - 开发你的机器学习工作流的全部內容，希望文章能夠幫你解決所遇到的問題。
- 上一篇： 云原生生态周报 Vol. 12 | K8
- 下一篇： 使用函数计算三步实现深度学习 AI 推理