如何基于流水线函数通过编程方式创建Kubeflow recurring run定时任务

方案1：直接编译输出JSON格式的Pipeline Spec #

KFP自带的编译器原生支持直接生成JSON格式的产物，仅需修改package_path的后缀为.json即可，无需额外转换操作：

from kfp import compiler
# 直接指定输出为JSON格式文件
compiler.Compiler().compile(
    pipeline_func=my_pipeline,
    package_path='pipelines/my_pipeline.json'
)

from kfp.v2.google.client import AIPlatformClient
api_client = AIPlatformClient(project_id='你的项目ID',
                             region='你的部署区域')

api_client.create_schedule_from_job_spec(
    job_spec_path='pipelines/my_pipeline.json', # 直接传入JSON文件路径
    schedule='* * * * *',
    time_zone='UTC',
    parameter_values=arguments
)

方案2：使用KFP原生客户端的create_recurring_run方法 #

如果是原生部署的Kubeflow，更推荐使用KFP官方客户端的定时创建接口，直接兼容YAML、TGZ格式的Pipeline Spec，也支持直接关联已上传到服务端的Pipeline ID，灵活度更高：

import kfp

# 初始化KFP客户端，填写你的Kubeflow访问端点
client = kfp.Client(host='https://你的-kfp访问端点')

# 方式A：直接传入本地编译好的YAML/TGZ文件路径
recurring_run = client.create_recurring_run(
    experiment_id='你的实验ID',
    job_name='定时任务名称',
    cron_expression='* * * * *',
    pipeline_package_path='pipelines/my_pipeline.tgz', # 原生支持tgz/yaml格式
    params=arguments,
    enabled=True
)

# 方式B：关联已上传到KFP服务端的Pipeline ID，无需每次传递本地文件
recurring_run = client.create_recurring_run(
    experiment_id='你的实验ID',
    job_name='定时任务名称',
    cron_expression='* * * * *',
    pipeline_id='已上传的pipeline_id',
    version_id='可选的pipeline版本ID',
    params=arguments,
    enabled=True
)

方案3：手动转换YAML格式为JSON/字典调用 #

如果已经有编译好的YAML文件不想重新编译，可以手动做格式转换后调用接口，也可以直接传入解析后的字典对象，无需落地为本地JSON文件：

import yaml
from kfp.v2.google.client import AIPlatformClient

# 加载本地YAML格式的pipeline spec
with open('my_pipeline.yaml', 'r', encoding='utf-8') as f:
    pipeline_spec = yaml.safe_load(f)

api_client = AIPlatformClient(project_id='你的项目ID',
                             region='你的部署区域')

# 直接传入解析后的字典对象，无需存储为本地JSON文件
api_client.create_schedule_from_job_spec(
    job_spec=pipeline_spec,
    schedule='* * * * *',
    time_zone='UTC',
    parameter_values=arguments
)

注意：以上方案适配KFP v1.8+ 及 v2.x版本，低版本客户端接口可能存在差异，请根据你的KFP服务版本调整对应参数。

内容的提问来源于stack exchange，提问作者jtlz2