使用Kubeflow構(gòu)建機(jī)器學(xué)習(xí)流水線

在此前的文章中，我已經(jīng)向你介紹了Kubeflow，這是一個(gè)為團(tuán)隊(duì)設(shè)置的機(jī)器學(xué)習(xí)平臺(tái)，需要構(gòu)建機(jī)器學(xué)習(xí)流水線。

在本文中，我們將了解如何采用現(xiàn)有的機(jī)器學(xué)習(xí)詳細(xì)并將其變成Kubeflow的機(jī)器學(xué)習(xí)流水線，進(jìn)而可以部署在Kubernetes上。在進(jìn)行本次練習(xí)的時(shí)候，請(qǐng)考慮你該如何將現(xiàn)有的機(jī)器學(xué)習(xí)項(xiàng)目轉(zhuǎn)換到Kubeflow上。

我將使用Fashion MNIST作為例子，因?yàn)樵诒敬尉毩?xí)中模型的復(fù)雜性并不是我們需要解決的主要目標(biāo)。對(duì)于這一簡(jiǎn)單的例子，我將流水線分為3個(gè)階段：

Git clone代碼庫(kù) 下載并重新處理訓(xùn)練和測(cè)試數(shù)據(jù) 訓(xùn)練評(píng)估

當(dāng)然，你可以根據(jù)自己的用例將流水線以任意形式拆分，并且可以隨意擴(kuò)展流水線。

獲取代碼

你可以從Github上獲取代碼：

% git clone https://github.com/benjamintanweihao/kubeflow-mnist.git

以下是我們用來(lái)創(chuàng)建流水線的完整清單。實(shí)際上，你的代碼很可能跨多個(gè)庫(kù)和文件。在我們的例子中，我們將代碼分為兩個(gè)腳本，preprocessing.py和train.py。

from tensorflow import keras import argparse import os import pickle def preprocess(data_dir: str): fashion_mnist = keras.datasets.fashion_mnist (train_images, train_labels), (test_images, test_labels) = fashion_mnist.load_data() train_imagestrain_images = train_images / 255.0 test_imagestest_images = test_images / 255.0 os.makedirs(data_dir, exist_ok=True) with open(os.path.join(data_dir, 'train_images.pickle'), 'wb') as f: pickle.dump(train_images, f) with open(os.path.join(data_dir, 'train_labels.pickle'), 'wb') as f: pickle.dump(train_labels, f) with open(os.path.join(data_dir, 'test_images.pickle'), 'wb') as f: pickle.dump(test_images, f) with open(os.path.join(data_dir, 'test_labels.pickle'), 'wb') as f: pickle.dump(test_labels, f) if __name__ == '__main__': parser = argparse.ArgumentParser(description='Kubeflow MNIST training script') parser.add_argument('--data_dir', help='path to images and labels.') args = parser.parse_args() preprocess(data_dir=args.data_dir)

處理腳本采用單個(gè)參數(shù)data_dir。它下載并預(yù)處理數(shù)據(jù)，并將pickled版本保存在data_dir中。在生產(chǎn)代碼中，這可能是TFRecords的存儲(chǔ)目錄。

train.py

import calendar import os import time import tensorflow as tf import pickle import argparse from tensorflow import keras from constants import PROJECT_ROOT def train(data_dir: str): # Training model = keras.Sequential([ keras.layers.Flatten(input_shape=(28, 28)), keras.layers.Dense(128, activation='relu'), keras.layers.Dense(10)]) model.compile(optimizer='adam', loss=tf.keras.losses.SparseCategoricalCrossentropy(from_logits=True), metrics=['accuracy']) with open(os.path.join(data_dir, 'train_images.pickle'), 'rb') as f: train_images = pickle.load(f) with open(os.path.join(data_dir, 'train_labels.pickle'), 'rb') as f: train_labels = pickle.load(f) model.fit(train_images, train_labels, epochs=10) with open(os.path.join(data_dir, 'test_images.pickle'), 'rb') as f: test_images = pickle.load(f) with open(os.path.join(data_dir, 'test_labels.pickle'), 'rb') as f: test_labels = pickle.load(f) # Evaluation test_loss, test_acc = model.evaluate(test_images, test_labels, verbose=2) print(f'Test Loss: {test_loss}') print(f'Test Acc: {test_acc}') # Save model ts = calendar.timegm(time.gmtime()) model_path = os.path.join(PROJECT_ROOT, f'mnist-{ts}.h5') tf.saved_model.save(model, model_path) with open(os.path.join(PROJECT_ROOT, 'output.txt'), 'w') as f: f.write(model_path) print(f'Model written to: {model_path}') if __name__ == '__main__': parser = argparse.ArgumentParser(description='Kubeflow FMNIST training script') parser.add_argument('--data_dir', help='path to images and labels.') args = parser.parse_args() train(data_dir=args.data_dir)

在train.py中，將建立模型，并使用data_dir指定訓(xùn)練和測(cè)試數(shù)據(jù)的位置。模型訓(xùn)練完畢并且開(kāi)始執(zhí)行評(píng)估后，將模型寫(xiě)入帶有時(shí)間戳的路徑。請(qǐng)注意，該路徑也已寫(xiě)入output.txt。稍后將對(duì)此進(jìn)行引用。

開(kāi)發(fā)Kubeflow流水線

為了開(kāi)始創(chuàng)建Kubeflow流水線，我們需要拉取一些依賴項(xiàng)。我準(zhǔn)備了一個(gè)environment.yml，其中包括了kfp 0.5.0、tensorflow以及其他所需的依賴項(xiàng)。

你需要安裝Conda，然后執(zhí)行以下步驟：

% conda env create -f environment.yml % source activate kubeflow-mnist % python preprocessing.py --data_dir=/path/to/data % python train.py --data_dir=/path/to/data

現(xiàn)在我們來(lái)回顧一下我們流水線中的幾個(gè)步驟：

Git clone代碼庫(kù) 下載并預(yù)處理訓(xùn)練和測(cè)試數(shù)據(jù) 訓(xùn)練并進(jìn)行評(píng)估

在我們開(kāi)始寫(xiě)代碼之前，需要從宏觀上了解Kubeflow流水線。

流水線由連接組件構(gòu)成。一個(gè)組件的輸出成為另一個(gè)組件的輸入，每個(gè)組件實(shí)際上都在容器中執(zhí)行（在本例中為Docker）。將發(fā)生的情況是，我們會(huì)執(zhí)行一個(gè)我們稍后將要指定的Docker鏡像，它包含了我們運(yùn)行preprocessing.py和train.py所需的一切。當(dāng)然，這兩個(gè)階段會(huì)有它們的組件。

我們還需要額外的一個(gè)鏡像以git clone項(xiàng)目。我們需要將項(xiàng)目bake到Docker鏡像，但在實(shí)際項(xiàng)目中，這可能會(huì)導(dǎo)致Docker鏡像的大小膨脹。

說(shuō)到Docker鏡像，我們應(yīng)該先創(chuàng)建一個(gè)。

Step0：創(chuàng)建一個(gè)Docker鏡像

如果你只是想進(jìn)行測(cè)試，那么這個(gè)步驟不是必須的，因?yàn)槲乙呀?jīng)在Docker Hub上準(zhǔn)備了一個(gè)鏡像。這是Dockerfile的全貌：

FROM tensorflow/tensorflow:1.14.0-gpu-py3 LABEL MAINTAINER "Benjamin Tan " SHELL ["/bin/bash", "-c"] # Set the locale RUN echo 'Acquire {http::Pipeline-Depth "0";};' >> /etc/apt/apt.conf RUN DEBIAN_FRONTEND="noninteractive" RUN apt-get update && apt-get -y install --no-install-recommends locales && locale-gen en_US.UTF-8 ENV LANG en_US.UTF-8 ENV LANGUAGE en_US:en ENV LC_ALL en_US.UTF-8 RUN apt-get install -y --no-install-recommends wget git python3-pip openssh-client python3-setuptools google-perftools && rm -rf /var/lib/apt/lists/* # install conda WORKDIR /tmp RUN wget --quiet https://repo.anaconda.com/miniconda/Miniconda3-4.7.12-Linux-x86_64.sh -O ~/miniconda.sh && /bin/bash ~/miniconda.sh -b -p /opt/conda && rm ~/miniconda.sh && ln -s /opt/conda/etc/profile.d/conda.sh /etc/profile.d/conda.sh && echo ". /opt/conda/etc/profile.d/conda.sh" >> ~/.bashrc # build conda environments COPY environment.yml /tmp/kubeflow-mnist/conda/ RUN /opt/conda/bin/conda update -n base -c defaults conda RUN /opt/conda/bin/conda env create -f /tmp/kubeflow-mnist/conda/environment.yml RUN /opt/conda/bin/conda clean -afy # Cleanup RUN rm -rf /workspace/{nvidia,docker}-examples && rm -rf /usr/local/nvidia-examples && rm /tmp/kubeflow-mnist/conda/environment.yml # switch to the conda environment RUN echo "conda activate kubeflow-mnist" >> ~/.bashrc ENV PATH /opt/conda/envs/kubeflow-mnist/bin:$PATH RUN /opt/conda/bin/activate kubeflow-mnist # make /bin/sh symlink to bash instead of dash: RUN echo "dash dash/sh boolean false" | debconf-set-selections && DEBIAN_FRONTEND=noninteractive dpkg-reconfigure dash # Set the new Allocator ENV LD_PRELOAD /usr/lib/x86_64-linux-gnu/libtcmalloc.so.

關(guān)于Dockerfile值得關(guān)注的重要一點(diǎn)是Conda環(huán)境是否設(shè)置完成并準(zhǔn)備就緒。要構(gòu)建鏡像：

% docker build -t your-user-name/kubeflow-mnist . -f Dockerfile % docker push your-user-name/kubeflow-mnist

那么，現(xiàn)在讓我們來(lái)創(chuàng)建第一個(gè)組件！

在pipeline.py中可以找到以下代碼片段。

Step1：Git Clone

在這一步中，我們將從遠(yuǎn)程的Git代碼庫(kù)中執(zhí)行一個(gè)git clone。特別是，我想要向你展示如何從私有倉(cāng)庫(kù)中進(jìn)行g(shù)it clone，因?yàn)檫@是大多數(shù)企業(yè)的項(xiàng)目所在的位置。當(dāng)然，這也是一個(gè)很好的機(jī)會(huì)來(lái)演示Rancher中一個(gè)很棒的功能，它能簡(jiǎn)單地添加諸如SSH密鑰之類的密鑰。

使用Rancher添加密鑰

訪問(wèn)Rancher界面。在左上角，選擇local，然后選擇二級(jí)菜單的Default：

然后，選擇Resources下的Secrets

你應(yīng)該看到一個(gè)密鑰的列表，它們正在被你剛剛選擇的集群所使用。點(diǎn)擊Add Secret：

使用你在下圖中所看到的值來(lái)填寫(xiě)該頁(yè)面。如果kubeflow沒(méi)有在命名空間欄下展示出來(lái)，你可以通過(guò)選擇Add to a new namespace并且輸入kubeflow簡(jiǎn)單地創(chuàng)建一個(gè)。

確保Scope僅是個(gè)命名空間。如果將Scope設(shè)置為所有命名空間，那么將使得在Default項(xiàng)目中的任意工作負(fù)載都能夠使用你的ssh密鑰。

在Secret Values中，key是id_rsa，值是id_rsa的內(nèi)容。完成之后，點(diǎn)擊Save。

如果一些進(jìn)展順利，你將會(huì)看到下圖的內(nèi)容?，F(xiàn)在你已經(jīng)成功地在kubeflow命名空間中添加了你的SSH密鑰，并且無(wú)需使用kubectl！

既然我們已經(jīng)添加了我們的SSH key，那么是時(shí)候回到代碼。我們?nèi)绾卫眯绿砑拥腟SH密鑰來(lái)訪問(wèn)私有g(shù)it倉(cāng)庫(kù)？

def git_clone_darkrai_op(repo_url: str): volume_op = dsl.VolumeOp( name="create pipeline volume", resource_name="pipeline-pvc", modes=["ReadWriteOnce"], size="3Gi" ) image = 'alpine/git:latest' commands = [ "mkdir ~/.ssh", "cp /etc/ssh-key/id_rsa ~/.ssh/id_rsa", "chmod 600 ~/.ssh/id_rsa", "ssh-keyscan bitbucket.org >> ~/.ssh/known_hosts", f"git clone {repo_url} {PROJECT_ROOT}", f"cd {PROJECT_ROOT}"] op = dsl.ContainerOp( name='git clone', imageimage=image, command=['sh'], arguments=['-c', ' && '.join(commands)], container_kwargs={'image_pull_policy': 'IfNotPresent'}, pvolumes={"/workspace": volume_op.volume} ) # Mount Git Secrets op.add_volume(V1Volume(name='ssh-key-volume', secret=V1SecretVolumeSource(secret_name='ssh-key-secret'))) op.add_volume_mount(V1VolumeMount(mount_path='/etc/ssh-key', name='ssh-key-volume', read_only=True)) return op

首先，創(chuàng)建一個(gè)Kubernetes volume，預(yù)定義大小為3Gi。其次，將image變量指定為我們將要使用的alpine/git Docker鏡像。之后是在Docker容器中執(zhí)行的命令列表。這些命令實(shí)質(zhì)上是設(shè)置SSH密鑰的，以便于流水線可以從私有倉(cāng)庫(kù)git clone，或者使用git://URL來(lái)代替 https://。

該函數(shù)的核心是下面一行，返回一個(gè)dsl.ContainerOp。

command和arguments指定了執(zhí)行鏡像之后需要執(zhí)行的命令。

最后一個(gè)變量十分有趣，是pvolumes，它是Pipeline Volumes簡(jiǎn)稱。它創(chuàng)建一個(gè)Kubernetes volume并允許流水線組件來(lái)共享單個(gè)存儲(chǔ)。該volume被掛載在/workspace上。那么這個(gè)組件要做的就是把倉(cāng)庫(kù)git clone到/workspace中。

使用Secrets

再次查看命令和復(fù)制SSH密鑰的位置。

流水線volume在哪里創(chuàng)建呢？當(dāng)我們將所有組件都整合到一個(gè)流水線中時(shí)，就會(huì)看到創(chuàng)建好的volume。我們?cè)?etc/ssh-key/上安裝secrets：

op.add_volume_mount(V1VolumeMount(mount_path='/etc/ssh-key', name='ssh-key-volume', read_only=True))

請(qǐng)記得我們將secret命名為ssh-key-secret：

op.add_volume(V1Volume(name='ssh-key-volume', secret=V1SecretVolumeSource(secret_name='ssh-key-secret')))

通過(guò)使用相同的volume名稱ssh-key-volume，我們可以把一切綁定在一起。

Step2：預(yù)處理

def preprocess_op(image: str, pvolume: PipelineVolume, data_dir: str): return dsl.ContainerOp( name='preprocessing', imageimage=image, command=[CONDA_PYTHON_CMD, f"{PROJECT_ROOT}/preprocessing.py"], arguments=["--data_dir", data_dir], container_kwargs={'image_pull_policy': 'IfNotPresent'}, pvolumes={"/workspace": pvolume} )

正如你所看到的， 預(yù)處理步驟看起來(lái)十分相似。

image指向我們?cè)赟tep0中創(chuàng)建的Docker鏡像。

這里的command使用指定的conda python簡(jiǎn)單地執(zhí)行了preprocessing.py腳本。變量data_dir被用于執(zhí)行preprocessing.py腳本。

在這一步驟中pvolume將在/workspace里有倉(cāng)庫(kù)，這意味著我們所有的腳本在這一階段都是可用的。并且在這一步中預(yù)處理數(shù)據(jù)會(huì)存儲(chǔ)在/workspace下的data_dir中。

Step3：訓(xùn)練和評(píng)估

def train_and_eval_op(image: str, pvolume: PipelineVolume, data_dir: str, ): return dsl.ContainerOp( name='training and evaluation', imageimage=image, command=[CONDA_PYTHON_CMD, f"{PROJECT_ROOT}/train.py"], arguments=["--data_dir", data_dir], file_outputs={'output': f'{PROJECT_ROOT}/output.txt'}, container_kwargs={'image_pull_policy': 'IfNotPresent'}, pvolumes={"/workspace": pvolume} )

最后，是時(shí)候進(jìn)行訓(xùn)練和評(píng)估這一步驟。這一步唯一的區(qū)別在于file_outputs變量。如果我們?cè)俅尾榭磘rain.py，則有以下代碼段：

with open(os.path.join(PROJECT_ROOT, 'output.txt'), 'w') as f: f.write(model_path) print(f'Model written to: {model_path}')

我們正在將模型路徑寫(xiě)入名為output.txt的文本文件中。通常，可以將其發(fā)送到下一個(gè)流水線組件，在這種情況下，該參數(shù)將包含模型的路徑。

將一切放在一起

要指定流水線，你需要使用dsl.pipeline來(lái)注釋流水線功能：

@dsl.pipeline( name='Fashion MNIST Training Pipeline', description='Fashion MNIST Training Pipeline to be executed on KubeFlow.' ) def training_pipeline(image: str = 'benjamintanweihao/kubeflow-mnist', repo_url: str = 'https://github.com/benjamintanweihao/kubeflow-mnist.git', data_dir: str = '/workspace'): git_clone = git_clone_darkrai_op(repo_urlrepo_url=repo_url) preprocess_data = preprocess_op(imageimage=image, pvolume=git_clone.pvolume, data_dirdata_dir=data_dir) _training_and_eval = train_and_eval_op(imageimage=image, pvolume=preprocess_data.pvolume, data_dirdata_dir=data_dir) if __name__ == '__main__': import kfp.compiler as compiler compiler.Compiler().compile(training_pipeline, __file__ + '.tar.gz')

還記得流水線組件的輸出是另一個(gè)組件的輸入嗎？在這里，git clone、container_op的pvolume將傳遞到preprocess_cp。

最后一部分將pipeline.py轉(zhuǎn)換為可執(zhí)行腳本。最后一步是編譯流水線：

% dsl-compile --py pipeline.py --output pipeline.tar.gz

上傳并執(zhí)行流水線

現(xiàn)在要進(jìn)行最有趣的部分啦！第一步，上傳流水線。點(diǎn)擊Upload a pipeline：

接下來(lái)，填寫(xiě)Pipeline Name和Pipeline Description，然后選擇Choose file并且指向pipeline.tar.gz以上傳流水線。

下一頁(yè)將會(huì)展示完整的流水線。我們所看到的是一個(gè)流水線的有向無(wú)環(huán)圖，在本例中這意味著依賴項(xiàng)會(huì)通往一個(gè)方向并且它不包含循環(huán)。點(diǎn)擊藍(lán)色按鈕Create run 以開(kāi)始訓(xùn)練。

大部分字段已經(jīng)已經(jīng)填寫(xiě)完畢。請(qǐng)注意，Run parameters與使用@ dsl.pipeline注釋的training_pipeline函數(shù)中指定的參數(shù)相同：

最后，當(dāng)你點(diǎn)擊藍(lán)色的Start按鈕時(shí)，整個(gè)流水線就開(kāi)始運(yùn)轉(zhuǎn)了！你點(diǎn)擊每個(gè)組件并查看日志就能夠知道發(fā)生了什么。當(dāng)整個(gè)流水線執(zhí)行完畢時(shí)，在所有組件的右方會(huì)有一個(gè)綠色的確認(rèn)標(biāo)志，如下所示：

結(jié)論

如果你從上一篇文章開(kāi)始就一直在關(guān)注，那么你應(yīng)該已經(jīng)安裝了Kubeflow，并且應(yīng)該能體會(huì)到大規(guī)模管理機(jī)器學(xué)習(xí)項(xiàng)目的復(fù)雜性。

在這篇文章中，我們先介紹了為Kubeflow準(zhǔn)備一個(gè)機(jī)器學(xué)習(xí)項(xiàng)目的過(guò)程，然后是構(gòu)建一個(gè)Kubeflow流水線，最后是使用Kubeflow接口上傳并執(zhí)行流水線。這種方法的奇妙之處在于，你的機(jī)器學(xué)習(xí)項(xiàng)目可以是簡(jiǎn)單的，也可以是復(fù)雜的，只要你愿意，你就可以使用相同的技術(shù)。

因?yàn)镵ubeflow使用Docker容器作為組件，你可以自由地加入任何你喜歡的工具。而且由于Kubeflow運(yùn)行在Kubernetes上，你可以讓Kubernetes處理機(jī)器學(xué)習(xí)工作負(fù)載的調(diào)度。

我們還了解了一個(gè)我喜歡的Rancher功能，它十分方便，可以輕松添加secrets。立刻，你就可以輕松地組織secrets（如SSH密鑰），并選擇將其分配到哪個(gè)命名空間，而無(wú)需為Base64編碼而煩惱。就像Rancher的應(yīng)用商店一樣，這些便利性使Kubernetes的工作更加愉快，更不容易出錯(cuò)。

當(dāng)然，Rancher提供的服務(wù)遠(yuǎn)不止這些，我鼓勵(lì)你自己去做一些探索。我相信你會(huì)偶然發(fā)現(xiàn)一些讓你大吃一驚的功能。Rancher作為一個(gè)開(kāi)源的企業(yè)級(jí)Kubernetes管理平臺(tái)，Run Kubernetes Everywhere一直是我們的愿景和宗旨。開(kāi)源和無(wú)廠商鎖定的特性，可以讓用戶輕松地在不同的基礎(chǔ)設(shè)施部署和使用Rancher。此外，Rancher極簡(jiǎn)的操作體驗(yàn)也可以讓用戶在不同的場(chǎng)景中利用Rancher提升效率，幫助開(kāi)發(fā)人員專注于創(chuàng)新，而無(wú)需在繁瑣的小事中浪費(fèi)精力。