KubeFlow-Pipeline 及 Argo 实现原理速析

摘要：Argo 是一个开源原生容器工作流引擎用于在 Kubernetes 上开发和运行应用程序。KubeFlow 的 Pipeline 子项目，由 Google 开源，其全面依赖 Argo 作为底层实现，并增强持久层来补充流程管理能力，同时通过 Python-SDK 来简化流程的编写。

Argo 是一个开源原生容器工作流引擎用于在 Kubernetes 上开发和运行应用程序。Argo Workflow 流程引擎，可以编排容器流程来执行业务逻辑，在 20 年 4 月 8 日进入 CNCF 孵化器组。

而 KubeFlow 的 Pipeline 子项目，由 Google 开源，其全面依赖 Argo 作为底层实现，并增强持久层来补充流程管理能力，同时通过 Python-SDK 来简化流程的编写。

1 Argo 流程引擎

Argo 的步骤间可以传递信息，即下一步（容器）可以获取上一步（容器）的结果。结果传递有 2 种：

1. 文件：上一步容器新生成的文件，会直接出现在下一步容器里面。

2. 信息：上一步的执行结果信息（如某文件内容），下一步也可以拿到。

下面我们就来解读一下，Argo 怎么实现“信息”在容器间的传递的，以及它和其他的流程引擎实现传递的区别。

1.2 文件怎么从上一个容器跑到下一个容器里的？

Argo 流程，可以指定 2 个步骤之间，传递结果文件（Artifact）。即假设流程为：A->B，那么 A 容器跑完，B 容器可以取得上一个容器的输出文件。

如下：A 容器生成一个 /tmp/hello_world.txt 文件，Argo 将这个文件，放到了 B 容器里面，并重命名为 /tmp/message 文件。

注意：流程上的每个步骤，都对应执行一个容器。 在 A 跑完后容器就退出了，然后才跑的 B（这时候已经没有 A 容器在运行了）。

所以 Argo 怎么把一个文件从 A 容器“拷贝”到 B 容器里面的？

1.2.1 容器间通过共享存储？（NO）

一般容器间共享文件，首先想到的都是：咱使用共享存储呀，大家都挂载同一个 PVC 不就行了。

确实共享存储可以实现容器共享文件，但是这里 Argo 可以：

（1）任意指定文件传递。（2）传递后文件可以改名字。

这 2 个是共享 Volume 做不到的，毕竟容器挂载目录得提前设定好，然后文件名大家看到的也是一样的。所以显然文件传递，不是通过共享 PVC 挂载实现的。

（Ps：不过 Argo 也在考虑这种实现方式，毕竟共享目录不需要任何额外 IO，透传效率更高。见：https://github.com/argoproj/argo/issues/1349）

1.2.2 通过管理面中转？（YES）

没有共享目录，那中转文件，只能是通过先取出来，再塞回去的方式喽。实际上 Argo 也确实这么做的，只是实现上还有些约束。

（1）“临时中转仓库”需要引入第三方软件（Minio）

（2）文件不能太大

（3）需要在用户容器侧，增加“代理”帮忙上传 &下载文件。

1.2.3 中转文件具体实现（docker cp）

现在我们打开 Argo 看看具体怎么实现的。因为你要取一个容器里面的文件，或者把一个文件放入一个容器，也不容易实现呢。

（1）小滑头 Argo 居给用户容器设置了一个 SideCar 容器，通过这个 SideCar 去读取用户的文件，然后上传到临时仓库。

（2）一个 Pod 里面的两个 Container，文件系统也是独立的，并不能直接取到另一个 Container 的文件。所以 Sidecar 容器为了取另一个容器里的文件，又把主机上面的 docker.sock 挂载进来了。这样就相当于拿到了主机 Root 权限，可以任意 cp 主机上任意容器里面的文件。

事实上，Sidecar 里面取文件的实现是：

docker cp -a 023ce:/tmp/hello_world.txt - | gzip > /argo/outputs/artifacts/hello-art.tgz

感觉稍微有点暴力。

1.2.4 中转实现的其他方式

实际上，通过 sidecar 容器提权到 root 权限，然后从用户的容器里面 copy 任意文件（即 docker cp 命令），只是 Argo 默认的实现。毕竟它自己也发现这样做安全上有点说不过去。

所以呢，它也留了其他方式去 copy 用户容器里面的文件。比如：kubectl 也是可以 cp 容器里面的文件的嘛。其他方式可参见：

https://github.com/argoproj/argo/blob/master/docs/workflow-executors.md

1.3 下一步容器怎么拿到上一步容器的结果？

Argo 流程，2 个步骤之间，除了传递文件，还可以传递结果信息（Information）。如：A->B，那么 A 容器跑完，B 容器可以取得上一个容器的一些 Information（不是整个文件）。

一般流程引擎透传信息，都是中转：

不过显然 Argo 自己没有存储 Information 的临时仓库，所以它得找个地方记录这些临时待中转的 information（虽然 Argo 找了 Minio 这个对象存储用来暂存中转文件，但是显然这货只能存文件，没有存 Metadata 元数据功能）。这里 Argo 又找了 Pod 里面的 Annotation 字段，当做临时中转仓库。先把信息记这里，下一步容器想要，就来这里取。

相信这里应该是有更好的实现方式的，这种把信息记录到 Annotation 的做法，约束比较大的（特别是 ETCD 的单个对象不能超过 1M 大小）。

可以考虑使用单独的 Configmap 来中转也可以。

2 KubeFlow-Pipeline 项目

KubeFlow-Pipeline 项目（简称 KFP），是 Kubeflow 社区开源的一个工作流项目，用于管理、部署端到端的机器学习工作流。KFP 提供了一个流程管理方案，方便将机器学习中的应用代码按照流水线的方式编排部署，形成可重复的工作流。

2.1 为什么要在 Argo 之上重新开发一套？

部署一套 Argo 很简单，启动一个 K8s-Controller 就行。可是部署一套 Kubeflow-Pipeline 系统就复杂多了，总共下来有 8 个组件。那是 Argo 什么地方不足，需要新开发一套 KFP，并搞这么复杂呢？主要的原因还在于 Argo 是基于 K8s 云原生这套理念，即 ETCD 充当“数据库”来运行的，导致约束比较大。

像：流程模板，历史执行记录，这些大量的信息很明显需要一个持久化层（数据库）来记录的，单纯依赖 ETCD 会有单条记录不能超过 1M，总记录大小不能超过 8G 的约束。

所以一个完整的流程引擎，包含一个数据库也都是很常规的。因此 KFP 在这一层做了较大的增强。

另外，在 ML 领域的用户界面层，KFP 也做了较多的用户体验改进。包括可以查看每一步的训练输出结果，直接通过 UI 进行可视化的图形展示。

https://www.kubeflow.org/docs/pipelines/sdk/output-viewer/

2.2 Kubeflow-Pipeline 后续演进点

见：https://github.com/kubeflow/pipelines/issues/3550

Dag 引擎组件的水平扩展（HPA）是其重要的一个特性，也是要成为一个成熟引擎所必要的能力。

当前 KFP 在稳定性以及组件的水平扩展上都还有待改进，因此商业使用还需要一段时间，这将是 KFP 未来的一个重要目标。

同时，使用权限过于高的 Sidecar 容器作为其实现步骤之间元数据传递的途径，也会是 KFP 生产级使用的一道门槛。或许在权限控制方面，KFP 需要思考一下其他规避途径，至少需要稍微增强一下。

概括一下：（1）水平扩展（HPA），（2）生产级可靠性，（3）安全增强。

3 流程引擎核心 &分层

3.1 DAG 核心

一个 DAG 流程引擎，核心代码也就 7 行大概能实现了：

例如下图示例：遍历发现步骤 D 没有依赖其他步骤，那么本次可以执行 D 步骤。

所以一般程序员一周时间总能开发一个“还能用”的流程引擎。但是完整的流程引擎却并不轻松

3.2 世界上为什么有这么多的流程引擎

DAG 基础核心非常简单，同时，各个领域想要做的事情却迥然不同。即使一个简单的步骤，大数据步骤说：“这一步要执行的 SQL 语句是 xxx”，而 K8s 任务步骤却说：“这一步执行需要的 Docker 镜像是 yyy”。

所以，各种各样的流程引擎就自然的出现了。

举几个例子：

AWS：Cloudformation 编排，Batch 服务，SageMaker-ML Pipeline，Data Pipeline

Azure：Pipeline 服务，ML Pipeline，Data Factory

Aliyun：函数 Pipeline 服务，ROS 资源编排，Batch 服务，PAI-Studio

大数据领域：Oozie，AirFlow

软件部署：Puppet，Chef，Ansible

基因分析：DNAnexus，NextFlow，Cromwell

每个领域总能找出一两个流程引擎，来控制谁先干活谁后干活。

总结一下：

（1）DAG 引擎核心很小

（2）各领域步骤的描述方式不一样

这就是为什么各个领域，总会有一个自己的流程引擎，而不像 K8s 能一统容器平台一样，出现一个能一统江湖的流程引擎。

3.3 DAG 引擎分层架构

成熟的流程引擎，应该有如下 4 层架构：

第一层：用户交互层。如：模板语法规则，Console 界面等

第二层：API 持久化层。如：模板记录，历史执行记录等

第三层：引擎实例层。如：能否水平扩容，流程是否有优先级等

第四层：驱动层。如：一个步骤能干什么活。跑一个容器还是跑一个 Spark 任务。

基本比较成熟的引擎都符合这种架构，例如 AirFlow 流程引擎，华为云的应用编排（AOS）引擎，数据湖工厂（DLF）引擎等都是如此。

目前 Argo 以及 Kubeflow-Pipeline 在引擎核心组件的水平扩展上，也即第三层引擎能力层稍有不足。同时其驱动层，目前也只能对接 K8s（即只能跑容器任务）。在选型的时候需要考虑进去。