4、流水线编排（上）：Pipeline 核心概念——组件（Component）与图（Graph）

好，咱们今天聊聊 Kubeflow Pipeline 里最基础、也最核心的两个概念：组件（Component）和图（Graph）。

说白了，一个 Pipeline 就是一张有向无环图（DAG）。图里的每个节点，就是一个组件。组件负责干活，图负责编排谁先干、谁后干、谁跟谁并行干。

我刚开始接触 Kubeflow 的时候，觉得这东西不就是把几个容器串起来吗？后来踩了不少坑才发现，这里面的门道还真不少。咱们一个一个说。

4.1 组件（Component）到底是什么？

组件，你可以理解成一个“黑盒子”。

它接收一些输入，执行一段逻辑，然后产生一些输出。在 Kubeflow 里，每个组件本质上就是一个 Docker 容器，外加一份描述文件。

这份描述文件，就是组件的“说明书”。它告诉 Pipeline：

• 这个组件叫什么名字
• 它需要哪些输入参数
• 它会输出哪些结果
• 它运行在哪个镜像里
• 它的启动命令是什么

举个例子，一个最简单的组件，可能只是打印一行日志：

# 一个简单的 Python 组件
def print_message(message: str) -> str:
    print(f"收到消息: {message}")
    return f"已处理: {message}"

但光有函数不行，你得把它“包装”成 Kubeflow 能识别的组件。我习惯用 kfp.components 来创建：

from kfp import components

# 将 Python 函数转换为组件
print_op = components.create_component_from_func(
    func=print_message,
    base_image="python:3.8",
    output_component_file="print_component.yaml"
)

嗯，这里要注意：base_image 一定要选对。我曾经因为用了 python:3.8-slim，结果里面缺了某个系统库，组件跑起来直接报错。后来我统一换成了 python:3.8 标准版，问题就解决了。

4.2 组件的输入与输出

组件的输入输出，是定义组件接口的关键。Kubeflow 支持多种类型：

类型	说明	示例
str	字符串	模型名称、文件路径
int	整数	批次大小、迭代次数
float	浮点数	学习率、阈值
bool	布尔值	是否启用缓存
list	列表	多个文件路径
dict	字典	配置参数
Dataset	数据集	Kubeflow 内置类型
Model	模型	Kubeflow 内置类型

我个人建议，能用简单类型就用简单类型。复杂类型（比如 Dataset）虽然功能强大，但调试起来比较麻烦。你想想看，如果组件 A 输出一个 Dataset，组件 B 去消费它，中间一旦出问题，你很难定位是 A 写错了，还是 B 读错了。

4.3 图（Graph）—— 组件的编排艺术

有了组件，下一步就是怎么把它们串起来。这就是图（Graph）要做的事。

在 Kubeflow 里，图是通过 Pipeline 函数来定义的。你写一个 Python 函数，函数里调用各个组件，组件之间的依赖关系就自动形成了。

看个例子：

from kfp import dsl

@dsl.pipeline(
    name="示例流水线",
    description="一个简单的数据预处理 + 训练流水线"
)
def my_pipeline(data_path: str, learning_rate: float = 0.01):
    # 第一步：加载数据
    load_task = load_data_op(data_path=data_path)
    
    # 第二步：数据清洗（依赖 load_task）
    clean_task = clean_data_op(
        input_data=load_task.outputs["output_data"]
    )
    
    # 第三步：特征工程（依赖 clean_task）
    feature_task = feature_engineering_op(
        clean_data=clean_task.outputs["cleaned_data"]
    )
    
    # 第四步：模型训练（依赖 feature_task）
    train_task = train_model_op(
        features=feature_task.outputs["features"],
        lr=learning_rate
    )

你看，这里没有显式地画图，但依赖关系已经通过 .outputs 隐式地建立起来了。Kubeflow 会自动解析这些依赖，生成一张 DAG。

为什么会这样？因为每个组件执行完后，会把输出存到 MinIO 或 S3 里。下游组件通过路径引用上游的输出。这种设计的好处是：

• 组件之间完全解耦
• 支持断点续跑（某个组件失败后，修复后可以从失败点继续）
• 方便并行执行（没有依赖关系的组件可以同时跑）

4.4 组件与图的关系——一张图说清楚

为了让你更直观地理解，我画了一张图：

这张图展示了一个典型的机器学习流水线。四个组件通过数据依赖连接起来，形成一个有向无环图。注意看，组件 1、2、3 是串行的，而组件 4 依赖组件 3 的输出。如果组件 2 失败了，组件 3 和 4 都不会执行。

4.5 组件的复用与共享

在实际项目中，你不可能每次都从头写组件。我习惯把常用的组件打包成 组件库，放在 Git 仓库里统一管理。

Kubeflow 支持从 YAML 文件加载组件：

# 从文件加载组件
load_data = components.load_component_from_file("components/load_data.yaml")
clean_data = components.load_component_from_file("components/clean_data.yaml")
train_model = components.load_component_from_file("components/train_model.yaml")

这样，团队里的每个人都可以复用这些组件。你想想看，如果每个人都在自己的 Pipeline 里写一遍数据加载逻辑，那维护起来得多痛苦。

4.6 组件的版本管理

组件也会迭代。今天的数据清洗逻辑，明天可能就要加一个新规则。怎么管理组件的版本？

我推荐两种方式：

1. 镜像 Tag 管理： 每次更新组件，都打一个新的 Docker 镜像 Tag，比如 clean_data:v1.0、clean_data:v1.1。
2. 组件 YAML 版本号： 在 YAML 文件里加一个 version 字段，方便追溯。

举个例子：

# components/clean_data.yaml
name: Clean Data
version: 1.1.0
description: 清洗原始数据，去除空值和异常值
inputs:
  - name: input_data
    type: String
    description: 输入数据路径
outputs:
  - name: cleaned_data
    type: String
    description: 清洗后数据路径
implementation:
  container:
    image: myregistry/clean_data:v1.1
    command: ["python", "clean.py"]
    args: [
      --input, {inputValue: input_data},
      --output, {outputPath: cleaned_data}
    ]

嗯，这里要注意：版本号一定要和镜像 Tag 对应上。我曾经因为版本号写错了，导致线上跑了一个旧版本的组件，排查了半天才发现是版本不匹配。

4.7 小结

组件和图，是 Kubeflow Pipeline 的基石。组件负责“做什么”，图负责“怎么做”。

• 组件是容器化的执行单元，有明确的输入输出
• 图是数据依赖的体现，决定了组件的执行顺序
• 组件可以复用、共享、版本管理

说白了，你只要把组件定义好，图就自动生成了。剩下的，交给 Kubeflow 去调度。

---

公众号:蓝海资料掘金营，微信deep3321