使用SageMaker真值合并容器时遭遇JSON Schema推断错误

报错原因 #

sagemaker-model-monitor-groundtruth-merger容器无法自动解析捕获数据或真值数据的JSON结构，必须手动指定数据Schema；同时代码中存在真值数据路径不匹配的问题，也会加剧Schema推断失败的概率。

解决步骤 #

1. 定义数据Schema #

根据你的捕获数据（endpointInput/endpointOutput）和真值数据的实际JSON结构，编写对应的Schema。例如：

• 捕获数据Schema示例：

  {
    "feature_columns": [
      {"name": "id", "type": "string"},
      {"name": "feature_1", "type": "float"},
      {"name": "feature_2", "type": "float"}
    ],
    "prediction_columns": [
      {"name": "prediction", "type": "int"}
    ]
  }
  

• 真值数据Schema示例：

  {
    "label_columns": [
      {"name": "id", "type": "string"},
      {"name": "true_label", "type": "int"}
    ]
  }
  

2. 修正环境变量与路径 #

修改代码中的两个关键部分：

• 统一真值数据的目标路径，确保ground_truth_source指向正确的目录；
• 在环境变量中添加手动指定的Schema。

修改后的完整代码：

import json
from datetime import datetime
from sagemaker.processing import Processor, ProcessingInput, ProcessingOutput
from sagemaker import get_model_monitor_container_uri

def run_merge_job_processor(
        region,
        instance_type,
        role,
        bucket_name,
        groundtruth_path,
        endpoint_input,
        merge_path,
        instance_count=1,
        ):
    # 修正真值数据路径：去掉datetime后缀，与ground_truth_source保持一致
    groundtruth_input_1 = ProcessingInput(input_name="groundtruth_input_1",
                              source="s3://{}/{}".format(bucket_name, groundtruth_path),
                              destination="/opt/ml/processing/groundtruth",
                              s3_data_type="S3Prefix",
                              s3_input_mode="File")
    
    endpoint_subpath = "/".join(endpoint_input.split("/")[3:])
    endpoint_input_1 = ProcessingInput(
                          input_name="endpoint_input_1",
                          source="s3://{}/{}".format(bucket_name, endpoint_input),
                          destination="/opt/ml/processing/input_data/{}".format(endpoint_subpath),
                          s3_data_type="S3Prefix",
                          s3_input_mode="File")
     
    output = ProcessingOutput(
                          output_name="result",
                          source="/opt/ml/processing/output",
                          destination=f"s3://{bucket_name}/{merge_path}")
    
    
    inputs = [ groundtruth_input_1, endpoint_input_1 ]
    outputs = [ output ]
    
    # 替换为你实际的字段结构
    dataset_schema = {
        "feature_columns": [
            {"name": "id", "type": "string"},
            {"name": "feature_1", "type": "float"},
            {"name": "feature_2", "type": "float"}
        ],
        "prediction_columns": [
            {"name": "prediction", "type": "int"}
        ]
    }
    
    ground_truth_schema = {
        "label_columns": [
            {"name": "id", "type": "string"},
            {"name": "true_label", "type": "int"}
        ]
    }
    
    env = {
        "dataset_format": "{\"sagemakerCaptureJson\": {\"captureIndexNames\": [\"endpointInput\",\"endpointOutput\"]}}",
        'dataset_source': '/opt/ml/processing/input_data',
        'ground_truth_source': '/opt/ml/processing/groundtruth',
        'output_path': '/opt/ml/processing/output',
        # 新增手动Schema配置
        'dataset_schema': json.dumps(dataset_schema),
        'ground_truth_schema': json.dumps(ground_truth_schema)
    }
    
    processor = Processor(image_uri=get_model_monitor_container_uri(region),
                          instance_count=instance_count,
                          instance_type=instance_type,
                          role=role,
                          env=env,
                          )
    
    return processor.run(
        inputs=inputs,
        outputs=outputs,
        wait=True
    )

3. 注意事项 #

• 确保Schema中的字段名、数据类型和你的实际JSON数据完全匹配；
• 如果通过id字段关联捕获数据与真值数据，需保证两边的id字段类型一致；
• 运行前确认S3数据路径正确，且SageMaker角色拥有对应bucket的读写权限。

内容的提问来源于stack exchange，提问作者SRIRAM SIDHARTHA R