按照上篇文章《解鎖云原生 AI 技能 | 在 Kubernetes 上構(gòu)建機(jī)器學(xué)習(xí)系統(tǒng)》搭建了一套 Kubeflow Pipelines 之后，我們一起小試牛刀，用一個真實(shí)的案例，學(xué)習(xí)如何開發(fā)一套基于 Kubeflow Pipelines 的機(jī)器學(xué)習(xí)工作流。

準(zhǔn)備工作

機(jī)器學(xué)習(xí)工作流是一個任務(wù)驅(qū)動的流程，同時也是數(shù)據(jù)驅(qū)動的流程，這里涉及到數(shù)據(jù)的導(dǎo)入和準(zhǔn)備、模型訓(xùn)練 Checkpoint 的導(dǎo)出評估、到最終模型的導(dǎo)出。這就需要分布式存儲作為傳輸?shù)拿浇?#xff0c;此處使用 NAS 作為分布式存儲。

• 創(chuàng)建分布式存儲，這里以 NAS 為例。此處 NFS_SERVER_IP 需要替換成真實(shí) NAS 服務(wù)器地址

創(chuàng)建阿里云 NAS 服務(wù)，可以參考文檔

需要在 NFS Server 中創(chuàng)建 /data

# mkdir -p /nfs # mount -t nfs -o vers=4.0 NFS_SERVER_IP:/ /nfs # mkdir -p /data # cd / # umount /nfs

創(chuàng)建對應(yīng)的 Persistent Volume

# cat nfs-pv.yaml apiVersion: v1 kind: PersistentVolume metadata:name: user-susanlabels:user-susan: pipelines spec:persistentVolumeReclaimPolicy: Retaincapacity:storage: 10GiaccessModes:- ReadWriteManynfs:server: NFS_SERVER_IPpath: "/data"# kubectl create -f nfs-pv.yaml

創(chuàng)建 Persistent Volume Claim

# cat nfs-pvc.yaml apiVersion: v1 kind: PersistentVolumeClaim metadata:name: user-susanannotations:description: "this is the mnist demo"owner: Tom spec:accessModes:- ReadWriteManyresources:requests:storage: 5Giselector:matchLabels:user-susan: pipelines # kubectl create -f nfs-pvc.yaml

開發(fā) Pipeline

由于 Kubeflow Pipelines 提供的例子都是依賴于 Google 的存儲服務(wù)，這導(dǎo)致國內(nèi)的用戶無法真正體驗(yàn) Pipelines 的能力。為此，阿里云容器服務(wù)團(tuán)隊(duì)提供了基于 NAS 存儲訓(xùn)練 MNIST 模型的例子，方便您在阿里云上使用和學(xué)習(xí) Kubeflow Pipelines。具體步驟分 3 步:

• (1) 下載數(shù)據(jù)
• (2) 利用 TensorFlow 進(jìn)行模型訓(xùn)練
• (3) 模型導(dǎo)出

在這 3 個步驟中，后一個步驟都依賴于前一個步驟而完成。
Kubeflow Pipelines 中可以用 Python 代碼描述這樣一個流程, 完整代碼可以查看 standalone_pipeline.py。
我們在例子中使用了基于開源項(xiàng)目 Arena 的 arena_op ，這是對于 Kubeflow 默認(rèn)的 container_op 封裝，它能夠?qū)崿F(xiàn)對于分布式訓(xùn)練 MPI 和 PS 模式的無縫銜接，另外也支持使用 GPU 和 RDMA 等異構(gòu)設(shè)備和分布式存儲的簡單接入，同時方便從 git 源同步代碼，是一個比較實(shí)用的工具 API。

@dsl.pipeline(name='pipeline to run jobs',description='shows how to run pipeline jobs.' ) def sample_pipeline(learning_rate='0.01',dropout='0.9',model_version='1',commit='f097575656f927d86d99dd64931042e1a9003cb2'):"""A pipeline for end to end machine learning workflow."""data=["user-susan:/training"]gpus=1 # 1. prepare dataprepare_data = arena.standalone_job_op(name="prepare-data",image="byrnedo/alpine-curl",data=data,command="mkdir -p /training/dataset/mnist && \cd /training/dataset/mnist && \curl -O https://code.aliyun.com/xiaozhou/tensorflow-sample-code/raw/master/data/t10k-images-idx3-ubyte.gz && \curl -O https://code.aliyun.com/xiaozhou/tensorflow-sample-code/raw/master/data/t10k-labels-idx1-ubyte.gz && \curl -O https://code.aliyun.com/xiaozhou/tensorflow-sample-code/raw/master/data/train-images-idx3-ubyte.gz && \curl -O https://code.aliyun.com/xiaozhou/tensorflow-sample-code/raw/master/data/train-labels-idx1-ubyte.gz")# 2. downalod source code and train the modelstrain = arena.standalone_job_op(name="train",image="tensorflow/tensorflow:1.11.0-gpu-py3",sync_source="https://code.aliyun.com/xiaozhou/tensorflow-sample-code.git",env=["GIT_SYNC_REV=%s" % (commit)],gpus=gpus,data=data,command='''echo %s;python code/tensorflow-sample-code/tfjob/docker/mnist/main.py \--max_steps 500 --data_dir /training/dataset/mnist \--log_dir /training/output/mnist --learning_rate %s \--dropout %s''' % (prepare_data.output, learning_rate, dropout),metrics=["Train-accuracy:PERCENTAGE"])# 3. export the modelexport_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))

Kubeflow Pipelines 會將上面的代碼轉(zhuǎn)化成一個有向無環(huán)圖 (DAG), 其中的每一個節(jié)點(diǎn)就是 Component (組件)，而 Component (組件)之間的連線代表它們之間的依賴關(guān)系。從 Pipelines UI 可以看到 DAG 圖:

首先具體理解一下數(shù)據(jù)準(zhǔn)備的部分，這里我們提供了 arena.standalone_job_op 的 Python API, 需要指定該步驟的名稱: name； 需要使用的容器鏡像: image； 要使用的數(shù)據(jù)以及其對應(yīng)到容器內(nèi)部的掛載目錄: data。
這里的 data 是一個數(shù)組格式, 如 data=[“user-susan:/training”]，表示可以掛載到多個數(shù)據(jù)。 其中 user-susan 是之前創(chuàng)建的 Persistent Volume Claim， 而 /training 為容器內(nèi)部的掛載目錄。

prepare_data = arena.standalone_job_op(name="prepare-data",image="byrnedo/alpine-curl",data=data,command="mkdir -p /training/dataset/mnist && \cd /training/dataset/mnist && \curl -O https://code.aliyun.com/xiaozhou/tensorflow-sample-code/raw/master/data/t10k-images-idx3-ubyte.gz && \curl -O https://code.aliyun.com/xiaozhou/tensorflow-sample-code/raw/master/data/t10k-labels-idx1-ubyte.gz && \curl -O https://code.aliyun.com/xiaozhou/tensorflow-sample-code/raw/master/data/train-images-idx3-ubyte.gz && \curl -O https://code.aliyun.com/xiaozhou/tensorflow-sample-code/raw/master/data/train-labels-idx1-ubyte.gz")

而上述步驟實(shí)際上是從指定地址利用 curl 下載數(shù)據(jù)到分布式存儲對應(yīng)的目錄 /training/dataset/mnist，請注意這里的 /training 為分布式存儲的根目錄，類似大家熟悉的根 mount 點(diǎn)；而 /training/dataset/mnist 是子目錄。其實(shí)后面的步驟可以通過使用同樣的根 mount 點(diǎn)，讀到數(shù)據(jù)，進(jìn)行運(yùn)算。
第二步是利用下載到分布式存儲的數(shù)據(jù),并通過 git 指定固定 commit id 下載代碼，并進(jìn)行模型訓(xùn)練。

train = arena.standalone_job_op(name="train",image="tensorflow/tensorflow:1.11.0-gpu-py3",sync_source="https://code.aliyun.com/xiaozhou/tensorflow-sample-code.git",env=["GIT_SYNC_REV=%s" % (commit)],gpus=gpus,data=data,command='''echo %s;python code/tensorflow-sample-code/tfjob/docker/mnist/main.py \--max_steps 500 --data_dir /training/dataset/mnist \--log_dir /training/output/mnist --learning_rate %s \--dropout %s''' % (prepare_data.output, learning_rate, dropout),metrics=["Train-accuracy:PERCENTAGE"])

可以看到這個步驟比數(shù)據(jù)準(zhǔn)備要相對復(fù)雜一點(diǎn)，除了和第一步驟中的 name， image, data 和 command 一樣需要指定之外，在模型訓(xùn)練步驟中，還需要指定:

• 獲取代碼的方式: 從可重現(xiàn)實(shí)驗(yàn)的角度來看，對于運(yùn)行試驗(yàn)代碼的追本溯源，是非常重要的一環(huán)。可以在 API 調(diào)用時指定 sync_source 的 git 代碼源，同時通過設(shè)定 env 中 GIT_SYNC_REV 指定訓(xùn)練代碼的 commit id；
• gpu: 默認(rèn)為 0，就是不使用 GPU；如果為大于 0 的整數(shù)值，就代表該步驟需要這個數(shù)量的 GPU 數(shù)；
• metrics: 同樣是從可重現(xiàn)和可比較的實(shí)驗(yàn)?zāi)康某霭l(fā)，用戶可以將需要的一系列指標(biāo)導(dǎo)出，并且通過 Pipelines UI 進(jìn)行直觀的顯示和比較。具體使用方法分為兩步：1. 在調(diào)用 API 時以數(shù)組的形式指定要收集指標(biāo)的 metrics name 和指標(biāo)的展示格式 PERCENTAGE 或者是 RAW，比如 metrics=["Train-accuracy:PERCENTAGE"]。 2. 由于 Pipelines 默認(rèn)會從 stdout 日志中收集指標(biāo)，你需要在真正運(yùn)行的模型代碼中輸出 {metrics name}={value} 或者 {metrics name}:{value}, 可以參考具體樣例代碼。

值得注意的是：

在本步驟中指定了和 prepare_data 相同的 data 參數(shù) [“user-susan:/training”]，就可以在訓(xùn)練代碼中讀到對應(yīng)的數(shù)據(jù)，比如 --data_dir /training/dataset/mnist。
另外由于該步驟依賴于 prepare_data，可以在方法中通過指定 prepare_data.output 表示兩個步驟的依賴關(guān)系。

最后 export_model 是基于 train 訓(xùn)練產(chǎn)生的 checkpoint，生成訓(xùn)練模型：

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 同步模型導(dǎo)出代碼并且利用共享目錄 /training/output/mnist 中的 checkpoint 執(zhí)行模型導(dǎo)出。
整個工作流程看起來還是很直觀的, 下面就可以定義一個 Python 方法將整個流程貫穿在一起：

@dsl.pipeline(name='pipeline to run jobs',description='shows how to run pipeline jobs.' ) def sample_pipeline(learning_rate='0.01',dropout='0.9',model_version='1',commit='f097575656f927d86d99dd64931042e1a9003cb2'):

@dsl.pipeline 是表示工作流的裝飾器，這個裝飾器中需要定義兩個屬性，分別是 name 和 description。
入口方法 sample_pipeline 中定義了 4 個參數(shù)： learning_rate, dropout, model_version 和 commit, 分別可以在上面的 train 和 export_model 階段使用。這里的參數(shù)的值實(shí)際上是 dsl.PipelineParam 類型，定義成 dsl.PipelineParam 的目的在于可以通過 Kubeflow Pipelines 的原生 UI 將其轉(zhuǎn)換成輸入表單，表單的關(guān)鍵字是參數(shù)名稱，而默認(rèn)值為參數(shù)的值。值得注意的是，這里的 dsl.PipelineParam 對應(yīng)值實(shí)際上只能是字符串和數(shù)字型；而數(shù)組和 map，以及自定義類型都是無法通過轉(zhuǎn)型進(jìn)行變換的。

實(shí)際上，這些參數(shù)都可以在用戶提交工作流時進(jìn)行覆蓋，以下就是提交工作流對應(yīng)的 UI：

提交 Pipeline

您可以在自己的 Kubernetes 內(nèi)將前面開發(fā)工作流的 Python DSL 提交到 Kubeflow Pipelines 服務(wù)中, 實(shí)際提交代碼很簡單:

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 代碼編譯成執(zhí)行引擎 (Argo) 識別的 DAG 配置文件；
通過 Kubeflow Pipeline 的客戶端創(chuàng)建或者找到已有的實(shí)驗(yàn)，并且提交之前編譯出的 DAG 配置文件。

在集群內(nèi)準(zhǔn)備一個 python3 的環(huán)境，并且安裝 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 的環(huán)境后,執(zhí)行如下命令，連續(xù)提交兩個不同參數(shù)的任務(wù)：

# 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

查看運(yùn)行結(jié)果

登錄到 Kubeflow Pipelines 的 UI: https://{pipeline地址}/pipeline/#/experiments， 比如：

https://11.124.285.171/pipeline/#/experiments

點(diǎn)擊 Compare runs 按鈕，可以比較兩個實(shí)驗(yàn)的輸入、花費(fèi)的時間和精度等一系列指標(biāo)。讓實(shí)驗(yàn)可追溯是讓實(shí)驗(yàn)可重現(xiàn)的第一步，而利用 Kubeflow Pipelines 本身的實(shí)驗(yàn)管理能力則是開啟實(shí)驗(yàn)可重現(xiàn)的第一步。

總結(jié)

實(shí)現(xiàn)一個可以運(yùn)行的 Kubeflow Pipeline 需要的步驟是：

構(gòu)建 Pipeline (流水線)中需要的最小執(zhí)行單元 Component (組件)，如果是利用原生定義的 dsl.container_ops, 需要構(gòu)建兩部分代碼：

• 構(gòu)建運(yùn)行時代碼：通常是為每個步驟構(gòu)建容器鏡像，作為 Pipelines 和真正執(zhí)行業(yè)務(wù)邏輯代碼之間的適配器。它所做的事情為獲取 Pipelines 上下文的輸入?yún)?shù)，調(diào)用業(yè)務(wù)邏輯代碼，并且將需要傳遞到下個步驟的輸出按照 Pipelines 的規(guī)則放到容器內(nèi)的指定位置，由底層工作流組件負(fù)責(zé)傳遞。 這樣產(chǎn)生的結(jié)果是運(yùn)行時代碼與業(yè)務(wù)邏輯代碼會耦合在一起。可以參考 Kubeflow Pipelines 的例子；
• 構(gòu)建客戶端代碼：這個步驟通常是長成下面的樣子, 熟悉 Kubernetes 的朋友會發(fā)現(xiàn)這個步驟實(shí)際上就是在編寫 Pod Spec：

container_op = dsl.ContainerOp(name=name,image='<train-image>',arguments=['--input_dir', input_dir,'--output_dir', output_dir,'--model_name', model_name,'--model_version', model_version,'--epochs', epochs],file_outputs={'output': '/output.txt'}) container_op.add_volume(k8s_client.V1Volume(host_path=k8s_client.V1HostPathVolumeSource(path=persistent_volume_path),name=persistent_volume_name)) container_op.add_volume_mount(k8s_client.V1VolumeMount(mount_path=persistent_volume_path,name=persistent_volume_name))

利用原生定義的 dsl.container_ops 的好處在于靈活，由于開放了和 Pipelines 的交互接口，用戶可以在 container_ops 這個層面做許多事情。但是它的問題在于：

• 復(fù)用度低。每個 Component 都需要構(gòu)建鏡像和開發(fā)運(yùn)行時代碼；
• 復(fù)雜度高。使用者需要了解 Kubernetes 的概念，比如 resource limit, PVC, node selector 等一系列概念；
• 支持分布式訓(xùn)練困難。由于 container_op 為單容器操作，如果需要支持分布式訓(xùn)練就需要在 container_ops 中提交和管理類似 TFJob 的任務(wù)。這里會帶來復(fù)雜度和安全性的雙重挑戰(zhàn)，復(fù)雜度比較好理解，安全性是說提交 TFJob 這類任務(wù)的權(quán)限會需要開放額外的權(quán)限給 Pipeline 的開發(fā)者。

另一種方式是使用 arena_op 這種可以重用的 Component API，它使用通用運(yùn)行時代碼，可以免去重復(fù)構(gòu)建運(yùn)行時代碼的工作；同時利用通用一套的 arena_op API 簡化用戶的使用；也支持 Parameter Server 和 MPI 等場景。建議您使用這種方式編譯 Pipelines。

將構(gòu)建好的 Component (組件)拼接成 Pipeline (流水線)；

將 Pipeline (流水線)編譯成 Argo 的執(zhí)行引擎 (Argo) 識別的 DAG 配置文件， 并提交 DAG 配置文件到 Kubeflow Pipelines, 利用 Kubeflow Pipelines 自身的 UI 查看流程結(jié)果。

總結(jié)

以上是生活随笔為你收集整理的解锁云原生 AI 技能 - 开发你的机器学习工作流的全部內(nèi)容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網(wǎng)站內(nèi)容還不錯，歡迎將生活随笔推薦給好友。