Vertex AI Pipeline调度：如何避免前序运行未结束时启动新任务？

你遇到的问题确实是Vertex AI原生调度的一个局限：max_concurrent_run_count=1的作用是限制同时运行的最大实例数，但它不会阻止cron按时间触发新运行——只要当前运行数没超过这个值（比如前序运行还在进行时，运行数是1，达到上限，理论上应该不会启动新的，但可能你的场景里有其他因素导致触发？不过不管怎样，原生调度确实没有直接支持“仅在前序运行完成后才触发下一次”的逻辑）。

目前没有内置的直接开关，但可以通过以下几种方式实现需求：

方法1：流水线完成后自动触发下一次运行 #

放弃cron调度，改为在流水线的最后一步添加一个自定义任务，当流水线成功（或失败后），调用Vertex AI API启动下一次运行。这样就能严格保证只有当前运行结束才会启动下一次。

示例思路：在你的流水线定义中，添加一个Python组件，代码逻辑大概是：

from kfp.v2.dsl import component
import google.cloud.aiplatform as aiplatform

@component(base_image="python:3.9")
def trigger_next_run(project: str, region: str, pipeline_template: str, params: dict):
    aiplatform.init(project=project, region=region)
    # 启动新的流水线运行
    pipeline_job = aiplatform.PipelineJob(
        display_name="your-pipeline",
        template_path=pipeline_template,
        parameter_values=params
    )
    pipeline_job.submit()

然后把这个组件作为流水线的最后一步，这样每次流水线完成后就会自动触发下一次运行。

方法2：用自定义调度器（Cloud Function + Cloud Scheduler） #

通过中间层来控制调度逻辑：用Cloud Scheduler按你的cron时间触发Cloud Function，在Cloud Function里先检查当前是否有活跃的流水线运行，只有没有活跃运行时才启动新的。

示例Cloud Function代码：

import google.cloud.aiplatform as aiplatform

def check_and_trigger_pipeline(event, context):
    # 初始化Vertex AI客户端
    aiplatform.init(project="your-project-id", region="your-region")
    
    # 查询当前处于RUNNING状态的目标流水线运行
    active_runs = aiplatform.PipelineJob.list(
        filter="display_name='your-pipeline-display-name' AND state='RUNNING'",
        order_by="create_time desc"
    )
    
    # 如果没有活跃运行，启动新的流水线
    if not active_runs:
        pipeline_job = aiplatform.PipelineJob(
            display_name="your-pipeline-display-name",
            template_path="gs://your-bucket/pipeline-template.json",
            parameter_values={"your-param": "your-value"}
        )
        pipeline_job.submit()

然后在Cloud Scheduler中配置cron表达式（比如*/1 * * * *），触发这个Cloud Function即可。

方法3：利用监控告警触发运行 #

通过Cloud Monitoring设置告警规则：当流水线运行进入SUCCEEDED或FAILED状态时，触发Cloud Function启动下一次运行。这种方式也能保证只有当前运行结束才会触发下一次。

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