Amazon SageMaker Pipeline报错：Properties类型无法JSON序列化

问题场景 #

搭建包含两步的SageMaker Pipeline：先做数据预处理生成scaled_data.csv、train.csv、test.csv，再用后两个文件训练RF模型。第二步运行时触发报错：

TypeError: Object of type Properties is not JSON serializable

相关代码 #

数据上传 #

# upload data from local path to default bucket with prefix raw_data
WORK_DIRECTORY = "data"

input_data = sagemaker_session.upload_data(
    path="{}/{}".format(WORK_DIRECTORY, "dataset.csv"),
    bucket=bucket,
    key_prefix="{}/{}".format(prefix, "input_data"),
)

第一步：数据预处理步骤 #

scaling_processor = SKLearnProcessor(
    framework_version=FRAMEWORK_VERSION,
    instance_type="ml.m5.4xlarge",
    instance_count=processing_instance_count,
    base_job_name="data-process",
    role=role,
    sagemaker_session=pipeline_session,
)

scaling_processor_args = scaling_processor.run(
    inputs=[
        ProcessingInput(source=input_data, destination="/opt/ml/processing/input"),
    ],
    outputs=[
        ProcessingOutput(output_name="scaled_data", source="/opt/ml/processing/output/scaled_data/"),
        ProcessingOutput(output_name="train", source="/opt/ml/processing/output/train/"),
        ProcessingOutput(output_name="test", source="/opt/ml/processing/output/test/")
    ],
    
    code="scripts/preprocess.py",
)

step_process = ProcessingStep(name="DataProcess", step_args=scaling_processor_args)

第二步：RF训练步骤（报错位置） #

estimator_cls = sagemaker.sklearn.SKLearn
FRAMEWORK_VERSION = "0.23-1"

rf_processor = FrameworkProcessor(
    estimator_cls,
    FRAMEWORK_VERSION,
    role = role,
    instance_count=1,
    instance_type='ml.m5.2xlarge',
    base_job_name='rf-modelling'
)

rf_processor_args = rf_processor.run(
    inputs=[
        ProcessingInput(source=step_process.properties.ProcessingOutputConfig.Outputs["train"].S3Output.S3Uri,
                        destination="/opt/ml/processing/input"),
        ProcessingInput(source=step_process.properties.ProcessingOutputConfig.Outputs["test"].S3Output.S3Uri,
                        destination="/opt/ml/processing/input"),
    ],
    outputs=[
        ProcessingOutput(output_name="rf_model",source = "/opt/ml/processing/output/")
    ],
    
    code="scripts/train.py",
)

step_train = ProcessingStep(name="RFTrain", step_args=rf_processor_args)

报错原因 #

直接通过step_process.properties.ProcessingOutputConfig.Outputs["train"].S3Output.S3Uri引用前一步输出是错误的：

• Properties是SageMaker Pipeline的动态运行时属性对象，在Pipeline定义阶段（序列化JSON提交给SageMaker），该对象还没有实际的S3 URI值，无法被序列化为JSON格式。
• 这种直接访问属性的方式只适用于运行时的Job内部，而非Pipeline定义阶段的跨步骤引用。

解决方案 #

使用SageMaker Pipeline提供的标准跨步骤输出引用方式，同时避免输入文件覆盖：

修改第二步的inputs部分代码：

rf_processor_args = rf_processor.run(
    inputs=[
        # 使用step_process的outputs属性引用前一步输出
        ProcessingInput(source=step_process.outputs["train"],
                        destination="/opt/ml/processing/input/train"),
        ProcessingInput(source=step_process.outputs["test"],
                        destination="/opt/ml/processing/input/test"),
    ],
    outputs=[
        ProcessingOutput(output_name="rf_model",source = "/opt/ml/processing/output/")
    ],
    
    code="scripts/train.py",
)

关键说明 #

1. 正确引用跨步骤输出：step_process.outputs["train"]是Pipeline定义阶段的合法占位符表达式，会在Pipeline运行时自动替换为实际的S3 URI，避免序列化问题。
2. 避免文件覆盖：将两个输入的destination设置为不同子目录（/train和/test），确保train.csv和test.csv不会互相覆盖，后续训练脚本可以分别从对应目录读取文件。

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