1 背景

近些年来，人工智能技术在自然语言处理、视觉图像和自动驾驶方面都取得不小的成就，无论是工业界还是学术界大家都在惊叹一个又一个的模型设计。但是对于真正做过算法工程落地的同学，在惊叹这些模型的同时，更多的是在忧虑如果快速且有效的将这些模型落地到业务中，并产生商业价值。正如Google  《Hidden Technical Debt in Machine Learning Systems》中说的，ML code仅是Machine Learning systems中的一小部分，像数据收集、特征抽取、配置管理、资源管理、模型部署、模型监控等同样十分的重要。

当我们看到一个典型的机器学习系统由这么多组件或子系统构成时，我们会进而想到另一个问题：这么多子系统应该如何高效的配合起来？

答案是机器学习工作流。 通过机器学习工作流，可以有效的将各个子系统串联起，每一个业务场景可以通过一个端到端的机器学习工作流来描述，同时通过工作流也可以追溯每一次模型产出或模型上线的元信息（例如数据、配置、base model等）。

在工业界，比较成熟的机器学习工作流是Google 的Vertex AI Pipeline和Amazon的Sagemaker Pipeline，大家如果感兴趣可以自行去他们的官网体验。

本文将要介绍的是开源机器学习工作流的解决方案Kubeflow Pipelines(注：Vertex AI Pipeline是基于Kubeflow Pipelines进行改造而来的）。

2 Kubeflow Pipelines

Kubeflow是一个基于云原生的Machine Learning Platform，旨于快速在kubernetes环境中构建一套开箱即用的机器学习平台。Kubeflow由多个子系统构成，覆盖了机器学习声明周期的全流程。

• 在训练阶段：Kubeflow针对不同机器学习框架提供了对应operator，方便用户在kubernetes环境中提交分布式训练任务;
• 在部署阶段：Kubeflow同样提供了多种的部署框架让用户根据自己的业务需求来选择;
• 在监控阶段：Kubeflow提供了Metadata和TensorBoard来对模型的Artifact、血缘和性能进行分析;

针对于机器学习上述阶段，Kubeflow提供了Kubeflow Pipeline工作流平台，用于将机器学习的各个阶段进行串联，同时提供了下述能力：

• 任务编排：Kubeflow Pipeline通过argo提供workflow的能力，能够实现丰富多样的DAG 工作流，用户可以根据的业务需求定义、管理和复用自己工作流；
• 实验管理：Kubeflow Pipeline通过Experiments的能力，能够展示和对比不同实验参数（例如：模型超参）下Pipeline的运行结果，用户可以根据结果来对工作流任务进行调优；
• 模型追溯：Kubeflow Pipeline通过Tracking的能力，能够记录每一次Pipeline运行中每个step的输入和输出信息，用户可以根据记录的内容进行问题排查或模型调优；

3 使用

3.1  编写Pipeline

Kubeflow Pipelines提供了Python的SDK让用户来快速构建符合自己业务场景的Pipeline。本节创建一个具有两个step的pipeline，第一个step读取参数，将内容输出到自己的output中，第二个step读取第一个step的output，然后输出的标准输出中。

pipeline代码逻辑如下

• 第5行~第10行：定义repeat_line step，读取参数并将结果输出到output path中
• 第14行~第19行：定义print_text step，从text_path中读取内容，并输出到stdout
• 第21行~23行：定义pipeline，repeat_line step 设置相应的参数，print_text step以repeat_line的output作为入参
• 第26行：编译pipeline，会生成一个print_repeating_lines_pipeline.yaml，本质上是一个argo 的workflow

importkfpfromkfp.componentsimportfunc_to_container_op, InputPath, OutputPathfromkfpimportcompiler@func_to_container_opdefrepeat_line(line: str, output_text_path: OutputPath(str), count: int=10):
'''Repeat the line specified number of times'''withopen(output_text_path, 'w') aswriter:
foriinrange(count):
writer.write(line+'\n')
# Reading bigger data@func_to_container_opdefprint_text(text_path: InputPath()): # The "text" input is untyped so that any data can be printed'''Print text'''withopen(text_path, 'r') asreader:
forlineinreader:
print(line, end='')
defprint_repeating_lines_pipeline():
repeat_lines_task=repeat_line(line='Hello', count=5000)
print_text(repeat_lines_task.output) # Don't forget .output !# Submit the pipeline for execution:compiler.Compiler().compile(pipeline_func=print_repeating_lines_pipeline, package_path='print_repeating_lines_pipeline.yaml')

$ python3.6 print_repeating_lines_pipeline.py
$ print_repeating_lines_pipeline.py  print_repeating_lines_pipeline.yaml

3.2  提交Pipeline

提交Pipeline有两种方式：

• 通过Kubeflow Pipelines UI提交
• 通过Kubeflow Pipelines SDK提交

通过Kubeflow Pipelines UI提交

• 在Pipelines 页面选择Upload pipelines，在详情页面选择上文生成的print_repeating_lines_pipeline.yaml文件

• 上传完pipeline后，选择create run来创建一个对应的Runs

• 跳转到runs的提交页面后，可以设置runs的相关信息，并点击start来进行提交

通过Kubeflow Pipeline SDK提交:

• 我们仅需要再上述代码的后面添加如下创建pipeline语句即可
• 然后执行对应的python脚本

kfp.Client().create_run_from_pipeline_func(print_repeating_lines_pipeline, arguments={})

3.3 查看Pipeline

在Kubeflow Pipeline UI的runs页面我们能够看到每次pipeline运行的detail list

并且可以点击具体的runs查看运行的详情

4 架构分析

了解完Kubeflow Pipelines的基本使用方式后，我们来看看Kubeflow Pipelines的服务架构。下面先给出一张Kubeflow Pipelines官网给出的架构图

整体上来讲，Kubeflow Pipelines由以下模块构成：

• Pipeline web server & Pipline Service：用来创建、管理和展示pipeline、experiments、runs和artifacts等信息；
• Pipeline Persistence Agent：用来watch pipeline执行的相关信息，并向信息写入到Mysql和mlmetada中；
• orchestration controllers：Kubeflow Pipelines所使用的controller，典型的如Argo Workflow controller用来执行workflow、Scheduler Workflow Controller用来执行定时任务；
• Artifact storeage：用于存储每次pipeline运行的input、output和日志等信息；

官方的这个架构图有一些抽象，初次看到这个架构图的时候感觉不是很清晰，因此这里我对Kubeflow Pipelines安装后的组件进行了进一步的分析。接下来我们以部署组件的维度来分析一下Kubeflow pipelines的架构。

下图是我对Kubeflow Pipelines安装后的组件按照功能的维度进行了划分

Kubeflow  Pipelines安装完共有14个组件，从功能上划分可以分为6类：

1. pipline server：处理来自SDK和UI的请求，用于Pipeline、Experiment、Run的的管理（创建、删除、查看、对比等）；

1. ml-pipeline-ui: 是使用typescript编写的nodejs server，它主要做两件事情：1）Kubeflow Pipeline UI静态页面的加载；2）为将前端页面的服务请求转发给ml-pipeline-server和mlmetadata
2. ml-pipeline-server：是通过mux、grpc和grpc-gateway构成的一个能够同时处理grpc和http请求的server，它主要用来响应Kubeflow Pipeline UI和Kubeflow Pipeline SDK发送的请求；
3. ml-pipeline-persistence-agent: 是一个k8s的controller，监听workflow和scheduler workflow两个对象，并将监听到的信息汇报给ml-pipeline-server并存储到mysql中

2. controller：用于实现常规工作流和定时任务工作流的controller；

1. workflow controller: Kubeflow Pipelines直接使用了argo的workflow controller来实现工作流的step调度功能
2. scheduled workflow controller：用来执行周期性workflow的controller，其会监听workflow和scheduler worklfow两类对象，并判断对应的scheduler workflow是否到达下一个周期，如果是则会创建一个新的workflow；

3. storage：存储pipelines相关的元数据和artifact数据；
4. MLmeta：用于实现mlmeta功能的组件

1. metadata grpc server：是google开源的ml metadata，用于record和retrieve机器学习相关的metada data；其本质是一个grpc server，外界通过grpc client进行数据的写入和读取，metadata server则将数据写入到mysql中；
2. metadata envoy deployment：是一个proxy，Kubeflow Pipeline UI通过grpc-web访问metadata grpc server时，需要使用其进行协议的转换；
3. metadata writer: 是一个controller，其watch所有属于workflow的pod，并调用metadata的grpc接口将pod的Artifact信息写入到metadata grpc server中；

5. cache step：用于实现workflow中step缓存的功能的组件，在pipeline的执行过程中，当某个step的signature没有发生变化时，会直接使用history output。cache功能是通过k8s中的mutating admission webhook来实现的

1. cache-deploy-deployment：用来创建MutatingAdmissionWebhook，会match所有label为pipelines.kubeflow.org/cache_enabled: "true" 的create pod请求，并发送给cache-server.
2. cache-server：webhookserver会判断request的pod是否命中cache，如果命中则会对pod进行更改，以使得该pod不再执行真正的工作流任务。

6. viewer：用于可视化的展示step的结果

5.不足与改进

在调研和使用开源的Kubeflow Pipelines过程中，我们发现了以下几个问题，并进行了改进：

• 存储可用性没有保障：

• 问题：开源版本MySQL和MinIO均没有使用高可用方案，存在服务不稳定和数据丢失的风险
• 解法：对Kubeflow Pipelines进行更改，将MySQL替换为阿里云OSS，将MinIO替换为阿里云OSS

• 多租户依赖Kubeflow 完整安装

• 问题：开源版本多租户管理依赖Istio和kubeflow server，无法同阿里云账号体系打通
• 解法：通过云原生AI套件中的控制台模块对Kubeflow Pipelines进行代理访问，打通阿里云账号体系，并提供多租户能力

• Kubeflow Pipelines SDK过于基础

• 问题：开源SDK对于算法开发人员使用成本比较高，无法开箱即用
• 解法：提供kfp arena sdk，一方面提供开箱即用的component，例如pytorcjob、tfjob等；另一方面集成AI套件的原生能力，例如coscheduling

如果想体验生产级别的机器学习工作流，欢迎使用云原生AI套件新上的Kubeflow Pipelines，具体使用方法见：通过Kubeflow Pipelines创建工作流 。 使用过程中如果有任何疑问欢迎加入“云原生AI套件客户交流群”钉钉群沟通与交流。