如何在SageMaker Pipeline中正确获取LambdaStep的返回值？

要在后续步骤中正确引用LambdaStep的输出，需通过step_lambda.outputs属性访问，而非properties.Outputs或OutputParameters，具体修正如下：

核心修正逻辑 #

• LambdaStep的输出会被映射到step_lambda.outputs字典中，键为你定义LambdaOutput时指定的output_name；
• step_lambda.properties.Outputs返回的是不可序列化的Properties对象，会导致JSON序列化错误；
• OutputParameters并非LambdaStep的属性，仅适用于其他步骤（如ProcessingStep），因此会触发属性不存在的报错。

修正后的代码示例 #

# 保留原LambdaStep定义
from sagemaker.workflow.lambda_step import LambdaStep, Lambda, LambdaOutput, LambdaOutputTypeEnum

output_param_1 = LambdaOutput(output_name="status", output_type=LambdaOutputTypeEnum.Integer)
output_param_2 = LambdaOutput(output_name="file_name_a_c_drop", output_type=LambdaOutputTypeEnum.String)
output_param_3 = LambdaOutput(output_name="file_name_q_c_drop", output_type=LambdaOutputTypeEnum.String)

step_lambda = LambdaStep(
    name="ProcessingLambda",
    lambda_func=Lambda(
        function_arn="arn:aws:lambda:us-east-1:xxxxxxxx:function:xxxxx"
    ),
    inputs={
        "input_data": input_data,
        "input_file": trigger_file,
        "input_bucket": trigger_bucket
    },
    outputs = [
        output_param_1, output_param_2, output_param_3
    ]
)

# 后续ProcessingJob的inputs修正写法
from sagemaker.processing import ProcessingInput

inputs = [
    ProcessingInput(source=step_lambda.outputs["file_name_q_c_drop"], destination="/opt/ml/processing/input"),
    ProcessingInput(source=step_lambda.outputs["file_name_a_c_drop"], destination="/opt/ml/processing/input"),
]

额外注意事项 #

确保你的Lambda函数返回值为标准JSON格式，且键名与LambdaOutput的output_name完全匹配，示例返回格式如下：

{
    "status": 0,
    "file_name_a_c_drop": "s3://your-bath/path/file1.csv",
    "file_name_q_c_drop": "s3://your-bath/path/file2.csv"
}

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