SageMaker Pipeline调用estimator.fit报Join类型无法JSON序列化

为计算机视觉模型构建SageMaker Pipeline时，图像数据存储在S3存储桶中，已通过ScriptProcessor完成数据预处理且流程运行正常，在构建训练用Estimator阶段触发报错：
TypeError: Object of type Join is not JSON serializable

复现代码 #

from sagemaker.tensorflow import TensorFlow


output_config = preprocessing_job_description["ProcessingOutputConfig"]
for output in output_config["Outputs"]:
    if output["OutputName"] == "train_data":
        preprocessed_training_data = output["S3Output"]["S3Uri"]
    if output["OutputName"] == "valid_data":
        preprocessed_test_data = output["S3Output"]["S3Uri"]

s3_train = "s3://bucketname/image_data/train/"
s3_val = "s3://bucketname/image_data/val/"


tf_estimator = TensorFlow(entry_point="train.py",
                          sagemaker_session=sess,
                          role=role,
                          instance_count=1, 
                          instance_type="ml.m5.xlarge",
                          # output_path = "/opt/ml/processing/output",
                          model_dir="s3://bucketname/image_data/output",
                          py_version='py37',
                          framework_version='2.4', 
                          hyperparameters={'epochs': epochs,
                                           'learning_rate': learning_rate, 
                                           'train_batch_size': 64,
                                          },
                          metric_definitions=metrics_definitions,
                          script_mode=True,
                          max_run=7200 # 最长运行2小时，单位为秒
                         )

tf_estimator.fit({"train": preprocessed_training_data})

运行环境 #

• sagemaker 2.94.0
• Python3 Data Science kernel
• boto3 1.24.8

已尝试操作 #

• 调整Estimator各项入参
• 显式指定checkpoint_s3_uri参数
  以上操作均未解决问题，错误栈显示API请求JSON序列化阶段无法识别Join类型对象。

从preprocessing_job_description中提取的preprocessed_training_data不是普通字符串格式的S3 URI，而是SageMaker Pipeline内置的Join类型动态表达式对象。这类对象是Pipeline定义阶段用来做路径占位的特殊类型，只有Pipeline实际运行时才会被解析为真实的路径字符串，直接在Pipeline定义阶段将其传入fit()方法触发API请求时，内置的JSON序列化逻辑无法识别该类型，就会抛出对应报错。
本质错误是代码在Pipeline定义层直接读取预处理任务的输出配置，此时预处理任务还未实际执行，拿到的S3Uri本身就是表达式占位符，不是可直接序列化的真实路径。

不要在Pipeline定义流程中直接实例化原生Estimator后手动读取预处理任务的输出配置，改用SageMaker Pipeline原生的TrainingStep封装训练逻辑，直接引用预处理步骤的输出属性作为训练输入，Pipeline执行时会自动解析Join类型的表达式，不会触发序列化错误：

1. 导入Pipeline相关依赖

from sagemaker.workflow.pipeline import Pipeline
from sagemaker.workflow.steps import ProcessingStep, TrainingStep
from sagemaker.inputs import TrainingInput

2. 定义完ScriptProcessor对应的预处理步骤（假设命名为step_process）后，直接通过步骤的properties属性获取输出路径，不需要手动遍历ProcessingOutputConfig：

train_data_uri = step_process.properties.ProcessingOutputConfig.Outputs["train_data"].S3Output.S3Uri
valid_data_uri = step_process.properties.ProcessingOutputConfig.Outputs["valid_data"].S3Output.S3Uri

3. 保留原有TensorFlow Estimator的定义逻辑，用TrainingStep包裹训练流程，传入上面获取的路径作为输入，不要手动调用tf_estimator.fit()：

step_train = TrainingStep(
    name="CVModelTraining",
    estimator=tf_estimator,
    inputs={
        "train": TrainingInput(
            s3_data=train_data_uri,
            content_type="application/x-image" # 可根据实际图像数据格式调整
        ),
        "validation": TrainingInput(
            s3_data=valid_data_uri,
            content_type="application/x-image"
        )
    }
)

4. 将预处理步骤、训练步骤按依赖顺序加入Pipeline定义即可，Pipeline运行时会自动按顺序执行预处理、训练流程，自动解析动态路径表达式。

调试提示：如果需要在Notebook中交互式测试Estimator的训练逻辑，不要传入Pipeline步骤的输出属性，直接传入s3_train/s3_val这类写死的字符串格式真实S3路径即可，Join类型的表达式对象仅在Pipeline定义上下文中生效。

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