CURL创建Vertex Pipeline调度时如何指定GCS路径的Pipeline Spec？

不能直接在创建Vertex Pipeline调度的CURL请求里把GCS路径填到pipelineSpec字段——这个字段要求传入google.protobuf.Struct类型的内嵌Pipeline配置，而非字符串格式的GCS路径，这也是你收到400错误的原因。

要实现「每周拉取GCS中最新Pipeline Spec执行」且无需重建调度的需求，你需要通过Cloud Function/Cloud Run作为中间层来实现，具体步骤如下：

1. 编写中间服务（以Cloud Function为例） #

创建一个Cloud Function，核心逻辑是：

• 从指定GCS路径（gs://bucket_name/artifacts/pipeline.yaml）读取Pipeline配置文件
• 将YAML内容解析为google.protobuf.Struct格式
• 调用Vertex AI Pipeline的projects.locations.pipelineJobs.create API启动单次运行

示例代码（Python）：

import yaml
import google.auth
from google.cloud import aiplatform

def trigger_pipeline(event, context):
    # 初始化Vertex AI客户端
    credentials, project_id = google.auth.default()
    aiplatform.init(project=project_id, location="us-central1")

    # 从GCS读取Pipeline YAML
    storage_client = google.cloud.storage.Client()
    bucket = storage_client.bucket("bucket_name")
    blob = bucket.blob("artifacts/pipeline.yaml")
    pipeline_yaml = blob.download_as_text()
    pipeline_spec = yaml.safe_load(pipeline_yaml)

    # 启动Pipeline运行
    job = aiplatform.PipelineJob(
        display_name="weekly-pipeline-run",
        template_path=None,
        pipeline_spec=pipeline_spec,
        parameter_values={}  # 按需传入参数
    )
    job.submit()
    return "Pipeline triggered successfully"

2. 用CURL创建Cloud Scheduler作业 #

将Cloud Scheduler的触发目标指向上述Cloud Function的HTTP端点，每周执行一次：

curl -X POST "https://cloudscheduler.googleapis.com/v1/projects/[PROJECT_ID]/locations/[LOCATION]/jobs" \
  -H "Authorization: Bearer $(gcloud auth print-access-token)" \
  -H "Content-Type: application/json" \
  -d '{
    "name": "weekly-pipeline-trigger",
    "schedule": "0 0 * * 0",  # 每周日凌晨触发，按需调整Cron表达式
    "timeZone": "UTC",
    "httpTarget": {
      "uri": "https://[LOCATION]-[PROJECT_ID].cloudfunctions.net/trigger_pipeline",
      "httpMethod": "POST",
      "oidcToken": {
        "serviceAccountEmail": "[SERVICE_ACCOUNT]@[PROJECT_ID].iam.gserviceaccount.com"
      }
    }
  }'

为什么这种方式可行？ #

Vertex AI的内置Pipeline调度（PipelineSchedule资源）要求pipelineSpec必须内嵌到调度配置中，无法直接引用外部GCS路径，因此每次修改Spec都需要更新调度资源。而通过中间层服务，每次调度触发时都会动态拉取GCS中的最新配置，完全满足你无需重建调度的需求。

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