4、第一个Pipeline：Pipeline SDK安装、编写第一个组件（add）、编写Pipeline定义、编译与上传

好，咱们正式开始动手了。

前面聊了那么多概念，什么组件、什么DAG、什么运行环境……说实话，光看理论容易犯困。我个人习惯是，先跑通一个最简单的例子，哪怕它只是“1+1=2”，也比看十页文档管用。

这一章，我们就来写一个真正的Kubeflow Pipeline。它很简单，就一个加法组件。但你别小看它，这个骨架搭好了，后面再复杂的流程，无非是往这个框架里塞更多的组件而已。

4.1 安装Pipeline SDK

写Pipeline之前，得先把工具装上。Kubeflow Pipelines的SDK叫kfp，它是Python的一个包。你可以在你的开发机、Jupyter Notebook或者任何能跑Python的地方安装它。

pip install kfp --upgrade

嗯，就这么一行。但我建议你最好在一个干净的虚拟环境里装。为什么？因为kfp依赖的protobuf、requests这些库，版本有时候挺敏感的。我在项目中遇到过好几次，因为全局环境里装了一些乱七八糟的包，导致kfp安装后编译报错。所以，venv或conda搞起来，没坏处。

装完之后，验证一下：

python -c "import kfp; print(kfp.__version__)"

能看到版本号，比如2.0.0或1.8.22，就说明装好了。注意，Kubeflow Pipelines v2和v1的写法有些区别，我们这里以v2为主，因为它是未来的趋势。

4.2 编写第一个组件：add

组件是Pipeline的最小单元。说白了，它就是一个“输入→处理→输出”的黑盒子。我们写一个最简单的加法组件，输入两个数，输出它们的和。

在Kubeflow Pipelines v2中，定义一个组件非常优雅，用Python装饰器就行。你看：

from kfp import dsl

@dsl.component
def add(a: int, b: int) -> int:
    '''计算两个整数的和'''
    return a + b

就这么简单？对，就这么简单。

你想想看，以前写组件，你得自己写Dockerfile，自己处理输入输出文件的序列化，麻烦得很。现在一个@dsl.component装饰器，就把这些脏活累活全包了。它会自动把你的Python函数包装成一个容器化的组件，输入输出参数也会自动处理。

4.3 编写Pipeline定义

组件写好了，接下来得把它们串起来，形成一个工作流。这就是Pipeline定义。

Pipeline定义也是一个Python函数，同样用装饰器@dsl.pipeline来标记。在这个函数里，我们实例化组件，并指定它们之间的数据依赖关系。

@dsl.pipeline(
    name='my-first-pipeline',
    description='一个简单的加法Pipeline'
)
def my_pipeline(a: int = 1, b: int = 2):
    # 实例化组件
    add_task = add(a=a, b=b)
    # 这里只有一个组件，所以没有后续依赖
    # 如果有多个组件，可以这样写：add_task = add(a=a, b=b).after(previous_task)

注意看，my_pipeline函数有两个参数a和b，并且给了默认值。这些参数会成为Pipeline的输入参数，你在上传到Kubeflow Dashboard后，可以在运行前修改它们。

另外，add_task = add(a=a, b=b)这一行，看起来像是在调用函数，但实际上它是在构建一个计算图。它并没有真正执行加法，而是告诉Kubeflow：“嘿，这里有一个叫add的组件，它的输入是a和b，输出结果存到add_task里。” 真正的执行，要等到Pipeline被提交到Kubeflow集群后才会发生。

4.4 编译与上传

Pipeline定义写好了，但它只是一个Python函数。要让Kubeflow认识它，我们需要把它编译成一个YAML文件。这个YAML文件就是Pipeline的“蓝图”，里面描述了所有的组件、依赖关系、镜像等信息。

编译很简单，一行代码：

from kfp import compiler

compiler.Compiler().compile(
    pipeline_func=my_pipeline,
    package_path='my_first_pipeline.yaml'
)

运行后，你会在当前目录下看到一个my_first_pipeline.yaml文件。你可以打开看看，里面内容挺多的，但核心就是描述了我们刚才定义的add组件和my_pipeline的结构。

接下来就是上传。上传有两种方式：

• 通过Kubeflow Dashboard上传： 打开你的Kubeflow中央面板，找到“Pipelines”菜单，点击“Upload Pipeline”，选择刚才编译好的YAML文件，填个名字，搞定。
• 通过Python SDK上传： 如果你喜欢自动化，可以用kfp.Client来上传。

我一般用第二种方式，因为方便集成到CI/CD里。代码也很简单：

import kfp

client = kfp.Client(host='YOUR_KUBEFLOW_HOST')
pipeline = client.upload_pipeline(
    pipeline_package_path='my_first_pipeline.yaml',
    pipeline_name='my-first-pipeline'
)

上传成功后，你就可以在Dashboard上看到你的Pipeline了。点进去，点击“Create Run”，输入a和b的值（比如3和5），然后运行。等几分钟，如果一切顺利，你就能看到运行结果：8。

4.5 本章知识体系

为了让你更直观地理解这一章的核心逻辑，我画了一张图。它展示了从“安装SDK”到“上传运行”的完整流程。

这张图把整个流程串起来了。你跟着这个顺序走，基本不会出错。从安装SDK开始，到编写组件、定义Pipeline、编译、上传，最后运行。每一步都是下一步的前提。

4.6 避坑指南

最后，分享几个我踩过的坑，希望能帮你省点时间：

• 版本匹配问题： kfp SDK的版本要和你的Kubeflow后端版本匹配。比如Kubeflow 1.7对应kfp 1.8.x，Kubeflow 2.0对应kfp 2.x。版本不匹配，上传时会报奇怪的错误。
• 组件函数名不要用下划线开头： 比如def _add(a,b)，Kubeflow会把它当成私有函数，编译时可能忽略掉。我一开始就犯过这个错，编译出来的YAML文件里死活找不到组件。
• 默认参数类型要明确： 在Pipeline函数的参数里，最好都加上类型注解和默认值。比如a: int = 1，而不是a=1。否则Kubeflow Dashboard上显示的类型可能是str，导致输入校验失败。

好了，第一个Pipeline就讲到这里。你现在可以动手试试了。哪怕只是跑通一个加法，那种“我的代码在Kubeflow上跑起来了”的感觉，还是挺爽的。

---