使用Kubeflow pipelines

pipelines简介

pipelines是一个机器学习工作流的抽象概念，这个工作流可以小到函数的过程、也可以大到机器学习从数据加载、变换、清洗、特征构建、模型训练等多个环节。
在kubeflow中，该组件能以ui界面的方式记录、交互、反馈实验、任务和每一次运行。
pipelines各流程组件构建成功后，会依据事先定义好的组件依赖关系构建DAG（有向无环图）。
在pipelines构建各流程组件前，需要将对应流程的业务代码打包成docker镜像文件（kubeflow中运行业务代码均以容器的方式实现）
业务代码、各组件的关系可以参考如下的关系:

构建pipelines依赖kubeflow提供的SDK，目前官方文档中有比较详细几种构建组件和组件复用等方法，关于SDK的使用可以参考：pipelines sdk简介
目前提交运行pipelines有2种方法：
二者本质都是使用sdk编译pipelines组件

在notebook中使用sdk提交pipelines至服务中心，直接可以在ui中查看pipelines实验运行进度。
将pipelines组件打成zip包通过ui上传至服务中心，同样可以在ui查看实验运行进度。

kubeflow虽然开源了一段时间，但是一些功能与google云平台耦合依然较重，官方提供的例子在本地k8s集群中使用有诸多坑，也不方便新手理解。下面我将以tensorflow经典的手写数字识别算法为例，在本地k8s集群实践pipelines。

划分业务流程制作业务镜像

为了方便理解pipelines流程组件，我在这里将实验流程分成了数据加载、模型训练、测试数据预测三个环节（实际工作中你可以依据自己的喜好划分流程）
各流程的组件结构如下图所示：

在构建pipelines前先简单介绍一下三个环节的业务代码：
数据加载环节：
load_data.py

from __future__ import absolute_import, division, print_function, \unicode_literalsimport argparse
import numpy as np# load data
def load_data(path):with np.load(path) as f:x_train, y_train = f['x_train'], f['y_train']x_test, y_test = f['x_test'], f['y_test']return (x_train, y_train), (x_test, y_test)# do data transform
def transform(output_dir, file_name):x_train_name = 'x_train.npy'x_test_name = 'x_test.npy'y_train_name = 'y_train.npy'y_test_name = 'y_test.npy'(x_train, y_train), (x_test, y_test) = load_data(output_dir + file_name)print("### loading data done.")x_train, x_test = x_train / 255.0, x_test / 255.0np.save(output_dir + x_train_name, x_train)np.save(output_dir + x_test_name, x_test)np.save(output_dir + y_train_name, y_train)np.save(output_dir + y_test_name, y_test)print("### data transform done.")with open(output_dir + 'train_test_data.txt', 'w') as f:f.write(output_dir + x_train_name + ',')f.write(output_dir + x_test_name + ',')f.write(output_dir + y_train_name + ',')f.write(output_dir + y_test_name)print("### write train and test data name to: train_test_data.txt done.")def parse_arguments():"""Parse command line arguments."""parser = argparse.ArgumentParser(description='Kubeflow MNIST load data script')parser.add_argument('--data_dir', type=str, required=True, help='local file dir')parser.add_argument('--file_name', type=str, required=True, help='local file to be input')args = parser.parse_args()return argsdef run():args = parse_arguments()transform(args.data_dir, args.file_name)if __name__ == '__main__':run()

这个环节的业务代码主要实现了：
从本地读取源数据 -> 特征变换 -> 分割训练集 -> 存储数据集到本地

代码中使用argparse，用户可以通过rest api的方式传入路径和数据参数运行代码。
代码后续的运行将在容器中进行，在这里使用NFS挂载k8s集群路径作为文件存储的固定地址。
代码中将分割数据集的绝对路径写入train_test_data.txt 为的是方便后续环节引用，同时便于与后续环节构成流程关系。

测试数据预测环节：
predict.py

from __future__ import absolute_import, division, print_function, \unicode_literalsimport argparse
import numpy as np
import pandas as pd
import tensorflow as tfdef predict(output_dir, model_file, data_file):"""all file use absolute dir:param output_dir::param model_file: `model.txt` absolute dir:param data_file: `train_test_data.txt` absolute dir:return:"""with open(model_file, 'r') as f:line = f.readline()model = tf.keras.models.load_model(line)with open(data_file, 'r') as f:line = f.readline()data_list = line.split(',')with open(data_list[1], 'rb') as f:x_test = np.load(f)with open(data_list[3], 'rb') as f:y_test = np.load(f)pre = model.predict(x_test)model.evaluate(x_test, y_test)df = pd.DataFrame(data=pre,columns=["prob_0", "prob_1", "prob_2", "prob_3", "prob_4", "prob_5", "prob_6", "prob_7", "prob_8","prob_9"])y_real = pd.DataFrame(data=y_test, columns=["real_number"])result = pd.concat([df, y_real], axis=1)result.to_csv(output_dir + 'result.csv')print("### save predict result file: result.csv")with open(output_dir + 'result.txt', 'w') as f:f.write(output_dir + 'result.csv')print("### write result path and name to: result.txt done.")def parse_arguments():"""Parse command line arguments."""parser = argparse.ArgumentParser(description='Kubeflow MNIST predict model script')parser.add_argument('--data_dir', type=str, required=True, help='local file dir')parser.add_argument('--model_file', type=str, required=True, help='a file write trained model absolute dir')parser.add_argument('--data_file', type=str, required=True, help='la file write train and test data absolute dir')args = parser.parse_args()return argsdef run():args = parse_arguments()predict(args.data_dir, args.model_file, args.data_file)if __name__ == '__main__':run()

这个环节的业务代码主要实现了：
依据train环节生成的 model.txt文件获取模型文件地址 -> 测试数据集预测 -> 模型性能评估 -> 保留预测结果文件

代码中使用argparse，用户可以通过rest api的方式传入路径和数据参数运行代码。
使用模型预测测试数据集，预测结果存储到集群挂载卷

制作业务代码镜像

构建pipelines组件前需要将业务代码打包成docker镜像，在这里以 load_data 组件为例介绍一下打包业务代码镜像:
准备好dockerfile：包含业务代码、打包依赖的基础镜像、程序入口
Dockerfile

FROM tensorflow/tensorflow:1.14.0-py3
ADD load_data.py .
ENTRYPOINT ["python", "load_data.py"]

使用dockerfile打包业务代码镜像:
build_image

#!/usr/bin/env bashdocker build -t mnist-load_data:v0.0.1 .

tips：依赖的tensorflow基础镜像，大家去dockerhub不限版本随意拉取一个就好了

更多工程细节和流程结构参考本例github地址：https://github.com/liuweibin6566396837/kubeflow-examples/tree/master/mnist_stage

在notebook中提交pipelines

在notebook中可以使用pipelines python SDK实现编译pipelines同时将pipelines提交运行，构建pipelines组件代码如下：
client.py

import kfp
from kfp import compiler
import kfp.dsl as dsl
import kfp.notebook
import kfp.gcp as gcpclient = kfp.Client()
from kubernetes import client as k8s_clientEXPERIMENT_NAME = 'mnist_op'
exp = client.create_experiment(name=EXPERIMENT_NAME)class load_dataOp(dsl.ContainerOp):"""load raw data from tensorflow, do data transform"""def __init__(self, data_dir, file_name):super(load_dataOp, self).__init__(name='load_data',image='mnist-load_data:v0.0.1',arguments=['--file_name', file_name,'--data_dir', data_dir,],file_outputs={'data_file': data_dir + 'train_test_data.txt'})class trainOp(dsl.ContainerOp):"""train keras model"""def __init__(self, data_dir, data_file):super(trainOp, self).__init__(name='train',image='mnist-train:v0.0.1',arguments=['--data_dir', data_dir,'--data_file', data_file,],file_outputs={'model_file': data_dir + 'model.txt'})class predictOp(dsl.ContainerOp):"""get predict by trained model"""def __init__(self, data_dir, model_file, data_file):super(predictOp, self).__init__(name='predict',image='mnist-predict:v0.0.1',arguments=['--data_dir', data_dir,'--model_file', model_file,'--data_file', data_file],file_outputs={'result_file': data_dir + 'result.txt'})@dsl.pipeline(name='MnistStage',description='shows how to define dsl.Condition.'
)
def MnistTest():data_dir = '/DATA/nfs-data/kubeflow-pv1/'file_name = 'mnist.npz'load_data = load_dataOp(data_dir, file_name).add_volume(k8s_client.V1Volume(name='mnist-pv',nfs=k8s_client.V1NFSVolumeSource(path='/DATA/nfs-data/kubeflow-pv1/',server='10.5.188.249'))).add_volume_mount(k8s_client.V1VolumeMount(mount_path='/DATA/nfs-data/kubeflow-pv1/', name='mnist-pv'))train = trainOp(data_dir, load_data.outputs['data_file']).add_volume(k8s_client.V1Volume(name='mnist-pv',nfs=k8s_client.V1NFSVolumeSource(path='/DATA/nfs-data/kubeflow-pv1/',server='10.5.188.249'))).add_volume_mount(k8s_client.V1VolumeMount(mount_path='/DATA/nfs-data/kubeflow-pv1/', name='mnist-pv'))predict = predictOp(data_dir, train.outputs['model_file'], load_data.outputs['data_file']).add_volume(k8s_client.V1Volume(name='mnist-pv',nfs=k8s_client.V1NFSVolumeSource(path='/DATA/nfs-data/kubeflow-pv1/',server='10.5.188.249'))).add_volume_mount(k8s_client.V1VolumeMount(mount_path='/DATA/nfs-data/kubeflow-pv1/', name='mnist-pv'))compiler.Compiler().compile(MnistTest, 'mnist.tar.gz')
run = client.run_pipeline(exp.id, 'wbliu4', 'mnist.tar.gz')

在pipelines代码中完成了load_data、train、predict三个环节实例化，以此实现pipelines组件的构造。
在组件中需引用各自业务代码的镜像、明确业务代码需要的rest api 传入参数、业务代码生成的文件和路径
最后在 MnistTest 显示实现pipelines各个组件的关联：
例如在本例中，train组件需要引用load_data组件生成的 train_test_data.txt文件获取训练、测试数据集地址；
predict组件需要引用load_data组件生成的 train_test_data.txt文件获取测试数据集地址、train组件生成的model.txt文件获取模型文件的地址;
notebook中运行完代码后会生成一个pipelines实验链接，通过链接可以访问到实验:

查看DAG结构：

pipelines各组件运行结果：

至此使用notebook提交pipelines实验成功！！！

将pipelines打包上传至UI

将业务代码打包成镜像
编译压缩工作流程代码为.tar.gz文件
将压缩文件上传至kubeflow ui后，k8s集群将解压缩该压缩文件中的yaml配置文件，然后起一个pipeline工作流任务。
在pipelines的ui界面可以看到pipeline信息和运行状态。