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 

import kfp
from kfp.components import func_to_container_op, InputPath, OutputPath
from kfp import compiler

@func_to_container_op
def repeat_line(line: str, output_text_path: OutputPath(str), count: int = 10):
    '''Repeat the line specified number of times'''
    with open(output_text_path, 'w') as writer:
        for i in range(count):
            writer.write(line + '\n')


# Reading bigger data
@func_to_container_op
def print_text(text_path: InputPath()): # The "text" input is untyped so that any data can be printed
    '''Print text'''
    with open(text_path, 'r') as reader:
        for line in reader:
            print(line, end = '')

def print_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中

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

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

1. storage ：存储pipelines相关的元数据和artifact数据；
2. 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中；

1. 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不再执行真正的工作流任务。

1. 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创建工作流