如何从AWS SageMaker Pipeline步骤内部查询所属流水线执行信息？

方案1：读取作业自动标签（推荐） #

SageMaker Pipeline在触发处理作业时，会自动为作业添加核心关联标签，直接读取这些标签就能快速获取流水线执行信息：

• sagemaker:pipeline-execution-arn：对应作业所属的流水线执行ARN
• sagemaker:pipeline-name：对应流水线名称

使用boto3实现的代码示例：

import boto3

sagemaker_client = boto3.client('sagemaker')

# 替换为你的处理作业ARN
job_arn = "arn:aws:sagemaker:us-east-1:123456789012:processing-job/your-processing-job-name"

# 获取作业标签
tag_response = sagemaker_client.list_tags(ResourceArn=job_arn)
tag_map = {tag['Key']: tag['Value'] for tag in tag_response['Tags']}

# 提取流水线执行ARN并查询详情
pipeline_execution_arn = tag_map.get('sagemaker:pipeline-execution-arn')
if pipeline_execution_arn:
    execution_details = sagemaker_client.describe_pipeline_execution(
        PipelineExecutionArn=pipeline_execution_arn
    )
    print("流水线执行详情:", execution_details)

方案2：通过Lineage API查询（补充方案） #

若标签方式无法满足需求，可利用Lineage API的list_contexts接口，以作业ARN为关联资源查询流水线执行上下文：

import boto3

sagemaker_client = boto3.client('sagemaker')

job_arn = "arn:aws:sagemaker:us-east-1:123456789012:processing-job/your-processing-job-name"

# 查询关联的PipelineExecution类型上下文
context_response = sagemaker_client.list_contexts(
    SourceUri=job_arn,
    ContextType='PipelineExecution'
)

if context_response['Contexts']:
    pipeline_execution_arn = context_response['Contexts'][0]['ContextArn']
    execution_details = sagemaker_client.describe_pipeline_execution(
        PipelineExecutionArn=pipeline_execution_arn
    )
    print("流水线执行详情:", execution_details)

避坑提示 #

不要采用遍历所有流水线、执行及步骤的方式筛选，这种方法会消耗大量API调用配额，且在流水线数量较多时延迟极高，完全是低效的冗余操作。

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