如何配置Step Functions等待SageMaker StartPipelineExecution完成后再执行下一步

有两种常用方式可以实现Step Functions等待SageMaker Pipeline执行完毕后再进入后续步骤，根据需求选择即可：

1. 同步等待（推荐） #

利用Step Functions与SageMaker的原生同步集成，无需手动轮询，Step Functions会自动等待Pipeline执行完成。

在状态机的JSON定义中，将Task的Resource指定为同步版本的ARN，并配置Pipeline参数：

"StartSageMakerPipeline": {
  "Type": "Task",
  "Resource": "arn:aws:states:::sagemaker:startPipelineExecution.sync",
  "Parameters": {
    "PipelineName": "你的Pipeline名称",
    "PipelineExecutionDisplayName": "StepFunctions触发的执行",
    "PipelineParameters": [
      {"Name": "参数1", "Value": "参数值1"}
    ]
  },
  "Next": "你的后续步骤名称"
}

配置完成后，这个Task会一直阻塞到Pipeline执行完成（成功或失败），之后自动进入下一个步骤。

2. 自定义轮询（灵活控制场景） #

如果需要自定义等待间隔、重试逻辑或状态判断逻辑，可以采用"启动Pipeline + 轮询状态"的组合方式：

步骤拆解： #

• 启动Pipeline：调用StartPipelineExecution获取执行ARN
• 等待间隔：设置固定时长的Wait状态，避免频繁查询
• 查询执行状态：调用DescribePipelineExecution获取当前状态
• 状态判断：根据返回的PipelineExecutionStatus决定后续流程：
  • 状态为Succeeded：进入后续步骤
  • 状态为Failed/Stopped：执行错误处理逻辑
  • 状态为InProgress：回到Wait状态继续轮询

示例状态机片段：

"StartPipeline": {
  "Type": "Task",
  "Resource": "arn:aws:states:::sagemaker:startPipelineExecution",
  "Parameters": {
    "PipelineName": "你的Pipeline名称"
  },
  "ResultPath": "$.pipelineExecution",
  "Next": "WaitForPipeline"
},
"WaitForPipeline": {
  "Type": "Wait",
  "Seconds": 30,
  "Next": "CheckPipelineStatus"
},
"CheckPipelineStatus": {
  "Type": "Task",
  "Resource": "arn:aws:states:::sagemaker:describePipelineExecution",
  "Parameters": {
    "PipelineExecutionArn": "$.pipelineExecution.PipelineExecutionArn"
  },
  "ResultPath": "$.pipelineStatus",
  "Next": "IsPipelineComplete"
},
"IsPipelineComplete": {
  "Type": "Choice",
  "Choices": [
    {
      "Variable": "$.pipelineStatus.PipelineExecutionStatus",
      "StringEquals": "Succeeded",
      "Next": "你的后续步骤名称"
    },
    {
      "Variable": "$.pipelineStatus.PipelineExecutionStatus",
      "StringEquals": "Failed",
      "Next": "Pipeline失败处理步骤"
    },
    {
      "Variable": "$.pipelineStatus.PipelineExecutionStatus",
      "StringEquals": "Stopped",
      "Next": "Pipeline终止处理步骤"
    }
  ],
  "Default": "WaitForPipeline"
}

注意事项： #

• 确保Step Functions的执行角色拥有/sagemaker:StartPipelineExecution、sagemaker:DescribePipelineExecution权限，以及SageMaker Pipeline执行所需的相关权限。
• 同步模式下要注意Step Functions的任务超时限制，若Pipeline执行时间过长，需调整状态机的超时配置。

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