如何在SageMaker转换任务Python代码中获取S3输出路径避免硬编码

要实现从Step Function配置的Transform Job中动态获取S3OutputPath并拆分存储桶名称，可按以下步骤操作：

1. 读取Transform Job环境变量获取任务名称 #

SageMaker Transform Job运行时会自动注入AWS_TRANSFORM_JOB_NAME环境变量，直接读取该变量即可拿到当前任务的唯一标识。

2. 通过SageMaker API拉取任务配置 #

使用boto3调用describe_transform_job接口，传入任务名称，就能获取包含TransformOutput在内的完整任务配置，从中提取目标S3OutputPath。

3. 解析S3路径提取存储桶名称 #

S3路径格式为s3://bucket-name/prefix，可手动拆分字符串，或用boto3内置工具函数解析。

完整代码示例 #

import os
import boto3
from botocore.exceptions import ClientError
from s3transfer.utils import parse_s3_url

def get_dynamic_s3_bucket():
    # 获取当前Transform Job名称
    transform_job_name = os.environ.get('AWS_TRANSFORM_JOB_NAME')
    if not transform_job_name:
        raise ValueError("代码未运行在SageMaker Transform Job环境中，无法获取任务名称")
    
    # 调用API获取任务配置
    sagemaker_client = boto3.client('sagemaker')
    try:
        job_details = sagemaker_client.describe_transform_job(TransformJobName=transform_job_name)
        s3_output_path = job_details['TransformOutput']['S3OutputPath']
    except ClientError as e:
        raise RuntimeError(f"获取Transform Job配置失败: {e.response['Error']['Message']}")
    
    # 解析S3路径获取桶名（两种方法二选一）
    # 方法一：手动拆分
    # if s3_output_path.startswith('s3://'):
    #     bucket_name = s3_output_path.split('/', 3)[2]
    # else:
    #     raise ValueError(f"无效的S3路径格式: {s3_output_path}")
    
    # 方法二：使用boto3工具函数
    bucket_name, _ = parse_s3_url(s3_output_path)
    
    return bucket_name

# 调用示例
target_bucket = get_dynamic_s3_bucket()
print(f"动态获取的存储桶名称: {target_bucket}")

注意事项 #

• 确保Transform Job绑定的IAM角色拥有sagemaker:DescribeTransformJob权限，否则会触发权限不足错误。
• 该逻辑仅在SageMaker Transform Job运行环境中生效，本地测试时需手动设置AWS_TRANSFORM_JOB_NAME环境变量并配置AWS凭证。

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