Kubeflow Pipelines v2 实战指南 — 用 KFP SDK 构建 ML 流水线

• 引言
• KFP v2 安装与基本概念
  • 安装
  • 核心概念
• 定义组件
  • 轻量级 Python 组件
  • 自定义 Docker 镜像组件
• 编写流水线
  • 基本流水线
  • 编译并运行流水线
  • 定期执行（Recurring Run）
• 高级模式
  • 并行执行（ParallelFor）
  • 缓存
  • 挂载卷
• CI/CD 集成
  • GitHub Actions + KFP
• 编译之后到底产出什么
• 提交执行并读懂一次 Run
• 从头到尾完整跑一次
• 组件比你想的要孤立得多
• 读一个失败 Run 的顺序
• 缓存默认是开着的
• 控制流和平台功能改名了
• 什么时候不该用 KFP
• 结语
• 参考资料
• 测验

引言

在把 ML 模型从实验搬到生产环境的过程中，可复现性、自动化、版本管理是必不可少的。Kubeflow Pipelines（KFP）v2 是一个在 Kubernetes 上定义和运行 ML 工作流的框架，仅凭 Python 装饰器就能组装出流水线。

本文将介绍 KFP v2 SDK 的核心功能，以及如何搭建实战流水线。

KFP v2 安装与基本概念

安装

pip install kfp==2.7.0

# 安装 Kubeflow Pipelines 后端（Kubernetes）
kubectl apply -k "github.com/kubeflow/pipelines/manifests/kustomize/env/platform-agnostic?ref=2.2.0"

# 端口转发
kubectl port-forward svc/ml-pipeline-ui -n kubeflow 8080:80

核心概念

# 1. Component：流水线中的一个工作单元（Python 函数）
# 2. Pipeline：由 Component 组成的 DAG（有向无环图）
# 3. Artifact：输入/输出数据（Dataset、Model、Metrics 等）
# 4. Run：流水线的一次执行
# 5. Experiment：多个 Run 的逻辑分组

定义组件

轻量级 Python 组件

from kfp import dsl
from kfp.dsl import (
    Dataset, Input, Output, Model, Metrics,
    ClassificationMetrics, component
)


@dsl.component(
    base_image="python:3.11-slim",
    packages_to_install=["pandas==2.1.4", "scikit-learn==1.4.0"]
)
def load_data(
    dataset_url: str,
    output_dataset: Output[Dataset]
):
    """数据加载组件"""
    import pandas as pd

    df = pd.read_csv(dataset_url)
    print(f"Loaded {len(df)} rows")

    # 保存到输出 artifact
    df.to_csv(output_dataset.path, index=False)
    output_dataset.metadata["num_rows"] = len(df)
    output_dataset.metadata["num_columns"] = len(df.columns)


@dsl.component(
    base_image="python:3.11-slim",
    packages_to_install=["pandas==2.1.4", "scikit-learn==1.4.0"]
)
def preprocess_data(
    input_dataset: Input[Dataset],
    train_dataset: Output[Dataset],
    test_dataset: Output[Dataset],
    test_size: float = 0.2
):
    """数据预处理与切分"""
    import pandas as pd
    from sklearn.model_selection import train_test_split

    df = pd.read_csv(input_dataset.path)

    # 预处理
    df = df.dropna()
    df = df.drop_duplicates()

    # 切分
    train_df, test_df = train_test_split(df, test_size=test_size, random_state=42)

    train_df.to_csv(train_dataset.path, index=False)
    test_df.to_csv(test_dataset.path, index=False)

    train_dataset.metadata["num_rows"] = len(train_df)
    test_dataset.metadata["num_rows"] = len(test_df)


@dsl.component(
    base_image="python:3.11-slim",
    packages_to_install=[
        "pandas==2.1.4", "scikit-learn==1.4.0",
        "joblib==1.3.2", "xgboost==2.0.3"
    ]
)
def train_model(
    train_dataset: Input[Dataset],
    model_output: Output[Model],
    metrics_output: Output[Metrics],
    n_estimators: int = 100,
    max_depth: int = 6,
    learning_rate: float = 0.1
):
    """模型训练"""
    import pandas as pd
    import joblib
    from xgboost import XGBClassifier
    from sklearn.model_selection import cross_val_score

    df = pd.read_csv(train_dataset.path)
    X = df.drop("target", axis=1)
    y = df["target"]

    # 训练
    model = XGBClassifier(
        n_estimators=n_estimators,
        max_depth=max_depth,
        learning_rate=learning_rate,
        random_state=42
    )
    model.fit(X, y)

    # 交叉验证
    cv_scores = cross_val_score(model, X, y, cv=5, scoring="accuracy")

    # 保存模型
    joblib.dump(model, model_output.path)
    model_output.metadata["framework"] = "xgboost"
    model_output.metadata["n_estimators"] = n_estimators

    # 记录指标
    metrics_output.log_metric("cv_accuracy_mean", float(cv_scores.mean()))
    metrics_output.log_metric("cv_accuracy_std", float(cv_scores.std()))
    metrics_output.log_metric("n_estimators", n_estimators)


@dsl.component(
    base_image="python:3.11-slim",
    packages_to_install=[
        "pandas==2.1.4", "scikit-learn==1.4.0",
        "joblib==1.3.2", "xgboost==2.0.3"
    ]
)
def evaluate_model(
    test_dataset: Input[Dataset],
    model_input: Input[Model],
    metrics_output: Output[ClassificationMetrics],
    eval_metrics: Output[Metrics]
) -> float:
    """模型评估"""
    import pandas as pd
    import joblib
    from sklearn.metrics import accuracy_score, classification_report

    df = pd.read_csv(test_dataset.path)
    X = df.drop("target", axis=1)
    y = df["target"]

    model = joblib.load(model_input.path)
    y_pred = model.predict(X)
    y_prob = model.predict_proba(X)

    accuracy = accuracy_score(y, y_pred)

    # 分类指标（混淆矩阵可视化）
    metrics_output.log_confusion_matrix(
        categories=["Class 0", "Class 1"],
        matrix=[[int(sum((y == 0) & (y_pred == 0))), int(sum((y == 0) & (y_pred == 1)))],
                [int(sum((y == 1) & (y_pred == 0))), int(sum((y == 1) & (y_pred == 1)))]]
    )

    eval_metrics.log_metric("test_accuracy", accuracy)

    return accuracy

自定义 Docker 镜像组件

@dsl.component(
    base_image="pytorch/pytorch:2.1.0-cuda12.1-cudnn8-runtime",
    packages_to_install=["transformers==4.37.0", "datasets==2.16.0"]
)
def finetune_llm(
    model_name: str,
    train_dataset: Input[Dataset],
    output_model: Output[Model],
    epochs: int = 3,
    batch_size: int = 8
):
    """LLM 微调（使用 GPU）"""
    from transformers import AutoModelForSequenceClassification, Trainer
    # ... 训练代码
    pass

编写流水线

基本流水线

@dsl.pipeline(
    name="ML Training Pipeline",
    description="数据加载 → 预处理 → 训练 → 评估流水线"
)
def ml_training_pipeline(
    dataset_url: str = "https://example.com/data.csv",
    test_size: float = 0.2,
    n_estimators: int = 100,
    max_depth: int = 6,
    learning_rate: float = 0.1,
    accuracy_threshold: float = 0.85
):
    # Step 1: 加载数据
    load_task = load_data(dataset_url=dataset_url)

    # Step 2: 预处理（load_task 完成后执行）
    preprocess_task = preprocess_data(
        input_dataset=load_task.outputs["output_dataset"],
        test_size=test_size
    )

    # Step 3: 模型训练
    train_task = train_model(
        train_dataset=preprocess_task.outputs["train_dataset"],
        n_estimators=n_estimators,
        max_depth=max_depth,
        learning_rate=learning_rate
    )
    # 设置资源限制
    train_task.set_cpu_limit("4")
    train_task.set_memory_limit("8Gi")

    # Step 4: 评估
    eval_task = evaluate_model(
        test_dataset=preprocess_task.outputs["test_dataset"],
        model_input=train_task.outputs["model_output"]
    )

    # Step 5: 条件部署
    with dsl.If(eval_task.output >= accuracy_threshold):
        deploy_task = deploy_model(
            model_input=train_task.outputs["model_output"],
            accuracy=eval_task.output
        )


@dsl.component(base_image="python:3.11-slim")
def deploy_model(
    model_input: Input[Model],
    accuracy: float
):
    """模型部署（满足条件时）"""
    print(f"Deploying model with accuracy: {accuracy:.4f}")
    print(f"Model path: {model_input.path}")
    # 实际部署逻辑（K8s Serving、BentoML 等）

编译并运行流水线

from kfp import compiler
from kfp.client import Client

# 1. 编译为 YAML
compiler.Compiler().compile(
    pipeline_func=ml_training_pipeline,
    package_path="ml_pipeline.yaml"
)

# 2. 提交到 KFP 服务器
client = Client(host="http://localhost:8080")

# 创建 Experiment
experiment = client.create_experiment(name="ml-experiments")

# 执行 Run
run = client.create_run_from_pipeline_func(
    ml_training_pipeline,
    experiment_name="ml-experiments",
    run_name="training-run-001",
    arguments={
        "dataset_url": "gs://my-bucket/data.csv",
        "n_estimators": 200,
        "max_depth": 8,
        "accuracy_threshold": 0.90
    }
)

print(f"Run ID: {run.run_id}")
print(f"Run URL: http://localhost:8080/#/runs/details/{run.run_id}")

定期执行（Recurring Run）

# 每天凌晨 2 点执行
client.create_recurring_run(
    experiment_id=experiment.experiment_id,
    job_name="daily-retraining",
    pipeline_func=ml_training_pipeline,
    cron_expression="0 2 * * *",
    max_concurrency=1,
    arguments={
        "dataset_url": "gs://my-bucket/latest-data.csv",
        "accuracy_threshold": 0.85
    }
)

高级模式

并行执行（ParallelFor）

@dsl.pipeline(name="Hyperparameter Search")
def hp_search_pipeline():
    # 定义超参数组合
    hp_configs = [
        {"n_estimators": 100, "max_depth": 4, "lr": 0.1},
        {"n_estimators": 200, "max_depth": 6, "lr": 0.05},
        {"n_estimators": 300, "max_depth": 8, "lr": 0.01},
    ]

    # 并行训练
    with dsl.ParallelFor(hp_configs) as config:
        train_task = train_model(
            train_dataset=load_task.outputs["output_dataset"],
            n_estimators=config.n_estimators,
            max_depth=config.max_depth,
            learning_rate=config.lr
        )

缓存

# 在组件级别禁用缓存
load_task = load_data(dataset_url=dataset_url)
load_task.set_caching_options(False)  # 始终重新执行

# 在流水线级别设置缓存
run = client.create_run_from_pipeline_func(
    ml_training_pipeline,
    enable_caching=True  # 输入相同时使用缓存
)

挂载卷

@dsl.component(base_image="python:3.11-slim")
def process_large_data(output_data: Output[Dataset]):
    """处理大规模数据"""
    pass

# 挂载 PVC
process_task = process_large_data()
process_task.add_pvolumes({
    "/mnt/data": dsl.PipelineVolume(pvc="data-pvc")
})

CI/CD 集成

GitHub Actions + KFP

# .github/workflows/ml-pipeline.yml
name: ML Pipeline CI/CD

on:
  push:
    branches: [main]
    paths:
      - 'pipelines/**'
      - 'components/**'

jobs:
  deploy-pipeline:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4

      - name: Set up Python
        uses: actions/setup-python@v5
        with:
          python-version: '3.11'

      - name: Install dependencies
        run: pip install kfp==2.7.0

      - name: Compile pipeline
        run: python pipelines/compile.py

      - name: Upload and run pipeline
        env:
          KFP_HOST: ${{ secrets.KFP_HOST }}
        run: |
          python -c "
          from kfp.client import Client
          client = Client(host='$KFP_HOST')
          client.upload_pipeline(
            pipeline_package_path='ml_pipeline.yaml',
            pipeline_name='ml-training-v2',
            description='Automated ML training pipeline'
          )
          "

编译之后到底产出什么

先说版本。从本节开始，API 以 kfp SDK 2.17.0 为准，并与 kfp-kubernetes 2.17.0 和 kfp-server-api 2.17.0 搭配使用。它需要 Python 3.9 以上，和上面安装代码块里的 2.7.0 并不相同。如果你找到的示例照抄过来跑不通，十有八九是 SDK 的版本差异。

API 参考把 Compiler 描述为「把用 KFP SDK DSL 编写的流水线编译成 YAML 流水线定义」，并明确把 package_path 定为「输出 YAML 文件的路径」。v2 的编译产物永远只有一个 YAML 文件。

from kfp import compiler

compiler.Compiler().compile(
    pipeline_func=ml_training_pipeline,
    package_path="ml_pipeline.yaml",  # 照搬文档的说法：「输出 YAML 文件的路径」
    pipeline_name="ml-training",
    pipeline_display_name="ML Training Pipeline",
    pipeline_parameters={"n_estimators": 200},
    type_check=True,
)
# 其余参数：kubernetes_manifest_options, kubernetes_manifest_format
# 注意：kfp_package_path 是 @dsl.component 的参数，不是 compile() 的

编译既不碰集群也不碰后端。所以它应该是 CI 里最先跑的那道检查。任务接错了或者类型对不上，都会在这里被拦下。想让 CI 步骤只有一行，用 CLI 更方便。

kfp dsl compile --py my_pipeline.py --output my_pipeline.yaml

kfp_package_path 长得像一个编译选项，很多人在这里绕远路。像 kubernetes_manifest_options 这种光看名字说不清用途的参数，请到你正在使用的版本的文档里确认准确的 API。

提交执行并读懂一次 Run

往上提交有两条路径。create_run_from_pipeline_func 接收流水线函数并在内部完成编译，create_run_from_pipeline_package 则接收你事先做好的 YAML。两者的参数集合是一样的。实际工作中推荐后者：CI 里产出的同一个 YAML，可以原样提交到预发和生产。

from kfp.client import Client

client = Client(host="http://localhost:8080", namespace="kubeflow")
experiment = client.create_experiment(name="ml-experiments")

# 两个函数共用的参数：arguments, run_name, experiment_name, namespace,
#   pipeline_root, enable_caching, cache_key, service_account, experiment_id
run = client.create_run_from_pipeline_package(
    pipeline_file="ml_pipeline.yaml",
    arguments={"n_estimators": 200, "accuracy_threshold": 0.90},
    run_name="training-run-001",
    experiment_id=experiment.experiment_id,
    enable_caching=True,
)

# 在 CI 里必须在这里阻塞，成功与否才能通过退出码传出去
client.wait_for_run_completion(run.run_id, timeout=3600, sleep_duration=5)

# 注册到目录并升版本
client.upload_pipeline("ml_pipeline.yaml", pipeline_name="ml-training-v2")
client.upload_pipeline_version("ml_pipeline.yaml", "v3", pipeline_name="ml-training-v2")

# 定期执行指向的是包或已注册的流水线，而不是函数
client.create_recurring_run(
    experiment_id=experiment.experiment_id,
    job_name="daily-retraining",
    pipeline_package_path="ml_pipeline.yaml",  # 或者 pipeline_id / version_id
    cron_expression="0 2 * * *",
    max_concurrency=1,
    no_catchup=True,      # 停机后恢复时，不把积压的调度一次性补跑完
    params={"accuracy_threshold": 0.85},
)

Client 的默认命名空间是 kubeflow，而 list_runs 的 page_size 默认是 10，所以只会返回十条。

定期执行有个坑。上面那段示例给 create_recurring_run 传了 pipeline_func，但确认过的签名里并没有这个参数。它必须指向编译好的包或者已注册流水线的 ID，而不是函数。

Run 一旦开始，图里的一个节点就是一个任务，一个任务就是一个 Pod。点开节点能看到日志、输入输出 Artifact，以及是否命中缓存。Pod 的命名规则和标签会随部署方式不同，请到你所用部署方式的文档里确认。

从头到尾完整跑一次

上面的示例需要外部 CSV 和 XGBoost，很难照原样跑起来。这里我们做一个复制过去就能直接提交的最小流水线。目的只有一个：把从编译到提交、再到 Artifact 出现在 UI 上的这条路径完整走一遍。

from typing import NamedTuple

from kfp import compiler, dsl
from kfp.client import Client


@dsl.component(base_image="python:3.11")
def make_split(n_rows: int, ratio: float) -> NamedTuple('outputs', train=int, test=int):
    from typing import NamedTuple

    outputs = NamedTuple('outputs', train=int, test=int)
    n_test = int(n_rows * ratio)
    return outputs(n_rows - n_test, n_test)


@dsl.component(base_image="python:3.11", packages_to_install=["scikit-learn==1.4.0"])
def fit_and_score(
    train_rows: int,
    model: dsl.Output[dsl.Model],
    metrics: dsl.Output[dsl.Metrics],
):
    import json

    from sklearn.dummy import DummyClassifier

    X = [[0], [1]] * train_rows
    y = [0, 1] * train_rows

    clf = DummyClassifier(strategy="most_frequent")
    clf.fit(X, y)
    accuracy = float(clf.score(X, y))

    with open(model.path, "w") as f:
        json.dump({"strategy": "most_frequent"}, f)
    model.metadata["framework"] = "sklearn"
    metrics.log_metric("accuracy", accuracy)


@dsl.pipeline(name="smoke-pipeline", description="compile to submit smoke test")
def smoke_pipeline(n_rows: int = 1000, ratio: float = 0.2):
    split = make_split(n_rows=n_rows, ratio=ratio)

    fit = fit_and_score(train_rows=split.outputs["train"])
    fit.set_memory_request("512Mi")
    fit.set_memory_limit("1Gi")
    fit.set_retry(num_retries=2)


compiler.Compiler().compile(smoke_pipeline, package_path="smoke_pipeline.yaml")

client = Client(host="http://localhost:8080")
run = client.create_run_from_pipeline_package(
    "smoke_pipeline.yaml",
    arguments={"n_rows": 1000},
    run_name="smoke-001",
    experiment_name="smoke",
)
print(run.run_id)
client.wait_for_run_completion(run.run_id, timeout=900)

# 参数映射：str -> string, int/float -> number, bool -> boolean,
#           list/dict -> object
# Artifact 类型：dsl.Artifact(system.Artifact), Dataset, Model, Metrics,
#   ClassificationMetrics, SlicedClassificationMetrics, HTML, Markdown
#   共有属性 .name .uri .path .metadata / Model 额外多出 .framework
# Metrics.log_metric(metric, value)
# ClassificationMetrics.log_roc_data_point(fpr, tpr, threshold), log_roc_curve(),
#   set_confusion_matrix_categories(), log_confusion_matrix_row(),
#   log_confusion_matrix_cell(), log_confusion_matrix(categories, matrix)
# dsl.InputPath / dsl.OutputPath 主要用于 Container Components，
#   Python Components 用返回值的类型注解做同样的事

跑起来大致是这个样子。字符串和 UI 的布局会随后端版本变化，所以只看每一项在不在即可。

1) 编译产物
   smoke_pipeline.yaml          <- 只有这一个文件

2) 提交之后的标准输出
   Run ID 以一行 UUID 打印出来

3) UI 中 Run 详情页的图
   make_split       Succeeded
   fit_and_score    Succeeded
     Metrics   标签页 : accuracy = 0.5
     Artifacts 标签页 : model (system.Model), metadata.framework = sklearn

4) 用同样的参数再提交一次
   make_split       Succeeded  (显示命中缓存)
   fit_and_score    Succeeded  (显示命中缓存)
   -> 执行时间缩短到几秒

验收一个新集群时，只看两件事就够了：第一次执行是否成功，第二次是否变快。两者合起来就是后端和 Artifact 存储已经接通的信号。

有多个输出的组件用 NamedTuple 声明，在下一个任务里通过 task.outputs['<output-key>'] 取出。在函数内部重新定义 NamedTuple 这段看着别扭的代码不是失误而是必需，原因下一节解释。

小的值会被序列化为参数，模型和数据集这类大的东西则成为 Artifact。按类型的映射和方法名都写在上面代码末尾的注释里。只挑一个说：记录 ROC 点的方法叫 log_roc_data_point，示例里时不时出现的 log_roc_reading 并不存在。

组件比你想的要孤立得多

这是最常踩的坑。文档为 Python 组件明确写了两条约束：函数的输入和输出必须带有有效的 KFP 类型注解，而且函数不能引用任何在其函数体之外定义的符号。

真正要命的是第二条。装饰器会把函数源码摘出来，放进容器里单独执行，所以模块顶部的 import、文件里某处的常量，在容器内都不存在。

# 不工作 - 引用了函数体外的符号
import pandas as pd

TARGET_COLUMN = "label"


@dsl.component(base_image="python:3.11")
def bad_component(data: dsl.Input[dsl.Dataset]) -> int:
    df = pd.read_csv(data.path)                # NameError: name 'pd' is not defined
    return int(df[TARGET_COLUMN].nunique())    # NameError: name 'TARGET_COLUMN' ...


# 能工作 - 所有符号都在函数体内
@dsl.component(base_image="python:3.11", packages_to_install=["pandas==2.1.4"])
def good_component(data: dsl.Input[dsl.Dataset], target_column: str = "label") -> int:
    import pandas as pd

    df = pd.read_csv(data.path)
    return int(df[target_column].nunique())

难受的地方在于编译能过。YAML 生成了，Run 也起来了，然后在集群上以 NameError 挂掉。评审规则只要一条：组件函数的第一行应该是 import，函数体里出现签名中没有的名字就打回。

packages_to_install 也有代价。文档说明这个列表会在任务每次执行时安装一遍。跑一百次，pip install 也跑一百次。替代方案是在构建期就把依赖烤进镜像的 Containerized Python Components。

# 只构建镜像，不推送 (本地确认用)
kfp component build src/ --component-filepattern my_component.py --no-push-image

# 一直推到镜像仓库 (CI 里用的形式)
kfp component build src/ --component-filepattern my_component.py --push-image

按 Containerized Python Components 文档，base_image 的默认值是 python:3.11。不过 Lightweight Python Components 页面上还写着 python:3.7，两个页面互相打架，所以永远显式写出来。在离线内网环境里，还需要一起用上 pip_index_urls、pip_trusted_hosts、install_kfp_package 和 use_venv。

读一个失败 Run 的顺序

Run 变红之后不要在 UI 上乱翻，按顺序来：找到失败的节点，只读日志最后三十行，然后分清这是 Python 异常还是 Pod 根本没起来。

• NameError 或 ModuleNotFoundError — 要么违反了隔离规则，要么缺包。编译通过在这里没有任何意义。
• 下一个任务找不到输入 — 你直接往 .uri 里写了。按文档的说法，.uri 是 Artifact 在存储中实际存在的位置，而 .path 提供便捷的本地文件系统访问。代码要写到 .path。
• Pod 卡在 Pending 不动 — request 是调度依据，limit 是上限。只把 request 调大就会永远 Pending，这在 Kubernetes 事件里能看到。
• GPU 任务起来了却没有 GPU — 看看 set_accelerator_type 和 set_accelerator_limit 是不是都设了。set_gpu_limit 在当前参考文档里已经没有了。
• 拉不到镜像 — 这是 Kubernetes 的通用行为而不是 KFP，去看集群侧的文档。
• 瞬间就结束了 — 那不是失败，是命中了缓存。

重试是 set_retry(num_retries, backoff_duration=None)，上游失败也要继续做的收尾工作是 ignore_upstream_failure()。想对整体成败做出反应，就用 dsl.ExitHandler 搭配 dsl.PipelineTaskFinalStatus，其中 state 是 SUCCEEDED、FAILED、CANCELLED 三者之一。

@dsl.component(base_image="python:3.11")
def notify(status: dsl.PipelineTaskFinalStatus):
    print("state:", status.state)
    if status.state == "FAILED":
        print("pipeline failed - send alert here")


@dsl.pipeline(name="pipeline-with-exit-handler")
def pipeline_with_exit_handler(n_rows: int = 1000):
    with dsl.ExitHandler(exit_task=notify()):
        split = make_split(n_rows=n_rows, ratio=0.2)

        fit = fit_and_score(train_rows=split.outputs["train"])
        fit.set_retry(num_retries=2)

夜间重训练悄无声息地失败，就是靠这个挡住的。backoff_duration 接受的值是什么格式没能确认，请到你正在使用的版本的文档里确认准确的 API。

缓存默认是开着的

文档明确写着，对所有组件而言缓存默认启用。你要做的不是打开它，它已经开着了，你只是在需要的时候把它关掉。

设置分三层。任务级是 set_caching_options(False)，Run 级是 enable_caching，而 Run 级会覆盖任务级。如果组件上已经关了却还是命中，去看 Run 的参数。

# 1) 任务级 - 只有这个任务总是重新执行
load_task = load_data(dataset_url=dataset_url)
load_task.set_caching_options(False)

# 2) Run 级 - 覆盖任务级的设置
run = client.create_run_from_pipeline_func(
    ml_training_pipeline,
    arguments={"dataset_url": "gs://my-bucket/data.csv"},
    enable_caching=False,
)

第三层是全局。编译标志或环境变量可以把默认值本身关掉，而环境变量必须在导入组件之前设置才会生效。

# 用编译标志关掉默认值
kfp dsl compile --py my_pipeline.py --output my_pipeline.yaml \
  --disable-execution-caching-by-default

# 或者用环境变量 (必须在 import 组件之前设置)
export KFP_DISABLE_EXECUTION_CACHING_BY_DEFAULT=true
python my_pipeline.py

缓存命中时，UI 上会出现一个绿色的云朵箭头图标。

从这里开始是推断而不是文档。缓存键由哪些部分组成没有文档说明，所以无法断言。不过从「组件和输入没变就会拿回上一次的输出」这个观察来看，让人意外的情况大多是改变的东西落在缓存键之外。外部存储桶里的数据被悄悄更新，或者 latest 这类浮动标签背后的镜像被换掉，都是典型例子。有疑问时，只把那个任务的缓存关掉再对比。

控制流和平台功能改名了

搬旧代码时第一个要确认的名字是 dsl.Condition。文档明确写着它已被功能等价的 dsl.If 取代并废弃。上面的流水线已经用的是 dsl.If，但公司仓库里 v2 早期的代码大概还留着它。

分支由 dsl.If、dsl.Elif、dsl.Else 组成一个整体，而当各分支要把不同任务的输出汇成一个时用 dsl.OneOf，这时必须存在 dsl.Else 分支。并行执行里最容易漏掉的参数，是 dsl.ParallelFor(items, name=None, parallelism=None) 的 parallelism。直接展开的话，有多少组合就会一次起多少个 Pod，集群一小就全部卡在 Pending。要汇总扇出的结果时用 dsl.Collected。

@dsl.pipeline(name="control-flow-example")
def control_flow_example(threshold: float = 0.85):
    # train_with_epochs, max_accuracy, promote_model, stage_model,
    # report_failure 假定都是你自己定义的组件

    # 用 parallelism 限制同时起来的 Pod 数量
    with dsl.ParallelFor(items=[1, 5, 10, 25], parallelism=2) as epochs:
        train_task = train_with_epochs(epochs=epochs)

    # 把扇出的结果汇成一个
    best = max_accuracy(models=dsl.Collected(train_task.outputs["model"]))

    with dsl.If(best.output >= threshold):
        promote_model(score=best.output)
    with dsl.Elif(best.output >= 0.70):
        stage_model(score=best.output)
    with dsl.Else():
        report_failure(score=best.output)

# PipelineTask 已确认的方法 (全部可以链式调用)
#   set_cpu_request, set_cpu_limit, set_memory_request, set_memory_limit,
#   set_accelerator_type, set_accelerator_limit, set_caching_options,
#   set_retry, set_env_variable, ignore_upstream_failure, after
#   set_gpu_limit 在当前 API 参考文档里已经没有

卷是单独的包。上面卷挂载示例里的 add_pvolumes 和 dsl.PipelineVolume 是 KFP v1 时代的写法，v2 里已确认的路径是用 pip install kfp[kubernetes] 安装的 kfp-kubernetes。

from kfp import dsl, kubernetes


@dsl.pipeline(name="pvc-example")
def pvc_example():
    pvc1 = kubernetes.CreatePVC(
        pvc_name_suffix='-my-pvc',
        access_modes=['ReadWriteMany'],
        size='5Gi',
        storage_class_name='standard',
    )

    task1 = producer()
    kubernetes.mount_pvc(task1, pvc_name=pvc1.outputs['name'], mount_path='/data')

    task2 = consumer().after(task1)
    kubernetes.mount_pvc(task2, pvc_name=pvc1.outputs['name'], mount_path='/data')

    # 连清理也在流水线内部完成
    kubernetes.DeletePVC(pvc_name=pvc1.outputs['name']).after(task2)


# 同一个包里已确认的其他功能
#   use_secret_as_env, use_secret_as_volume, use_config_map_as_env,
#   use_config_map_as_volume, add_ephemeral_volume, add_pod_label,
#   add_pod_annotation, use_field_path_as_env, set_timeout,
#   set_image_pull_policy, set_security_context, set_image_pull_secrets

节点选择、规避 taint 这些在 GPU 集群上经常需要的功能，其函数名没能确认，请到你正在使用的版本的文档里确认准确的 API。

什么时候不该用 KFP

KFP 不是一个轻量工具。跑一条流水线，需要 Kubernetes 集群、KFP 后端和对象存储全都活着。把它装到不需要这份可复现性和血缘追踪的活儿上，剩下的就只有成本。

• 一个 Python 脚本就能搞定的事 — 一旦拆成组件，各阶段之间的数据就要序列化并往返存储。几秒钟的作业，开销比作业本身还大。
• 只是在固定时刻重复同一件事 — 如果根本用不上血缘，cron 或 Kubernetes CronJob 就够了。
• 身边没有人能读懂 Kubernetes — KFP 的失败大多来自 Kubernetes。没有人能读集群，调试就成了算命。
• 团队已经在别的编排器上安顿下来 — 从既有工具里调用训练作业通常更便宜。
• 还处在探索阶段 — 每次都要构建镜像或者等 pip install 的那一刻，迭代速度就垮了。

反过来，KFP 真正创造价值的地方既窄又清楚：多个人跑同一份训练却得到不同结果的时候，以及需要回溯三个月前的模型是用什么数据做出来的时候。

结语

Kubeflow Pipelines v2 核心要点：

1. @dsl.component：将 Python 函数转换为容器化的组件
2. @dsl.pipeline：把组件连接成 DAG
3. Artifact 系统：用 Dataset、Model、Metrics 类型管理输入/输出
4. 条件/循环：用 dsl.If、dsl.ParallelFor 构建动态流水线
5. 缓存：相同输入时跳过重新执行，节省成本

真正吃掉时间的并不是这五条，而是隔离规则和默认开启的缓存。

参考资料

同一件事在不同页面上有时写法不一样，遇到这种情况，readthedocs 那边更接近真实的签名。

• dsl - 2026-08-16 确认。
• compiler - 2026-08-16 确认。
• client - 2026-08-16 确认。
• Containerized Python Components - 2026-08-16 确认。
• Artifacts - 2026-08-16 确认。
• Control flow - 2026-08-16 确认。
• Caching - 2026-08-16 确认。
• Platform-specific features - 2026-08-16 确认。

测验