SageMaker批量转换Job调用报422 Unprocessable Entity错误求助

健康检查/ping可正常通过，但执行预测调用时返回POST /invocations HTTP/1.1" 422 Unprocessable Entity，已尝试调整S3数据源路径、重构接口参数、修改CSV文件表头，问题仍未解决。以下是相关配置与代码：

Dockerfile #

FROM python:3.7
COPY requirements.txt /opt/program/requirements.txt 
RUN apt-get -y update && apt-get install -y --no-install-recommends \
wget \
nginx \
         
ca-certificates \
    
&& rm -rf /var/lib/apt/lists/*
RUN pip3 install --no-cache-dir -r /opt/program/requirements.txt
ENV AWS_DEFAULT_REGION=eu-east-1
ENV PYTHONUNBUFFERED=TRUE
ENV PYTHONDONTWRITEBYTECODE=TRUE
ENV PATH="/opt/program:${PATH}"
ENV MODEL_PATH="/opt/ml/model"

COPY ./model /opt/program
WORKDIR /opt/program

目录结构 #

└── <parent_folder>/
    ├── model/
    │   ├── serve
    │   └── predictor.py
    ├── Dockerfile
    └── Requirements

predictor.py代码 #

import os
import pickle
import io
from fastapi import FastAPI, HTTPException, Response, status
import pandas as pd

app = FastAPI()

model_path = os.environ['MODEL_PATH']

class ScoringService:
    model = None

    @classmethod
    def get_model(cls):
        if cls.model is None:
            with open(os.path.join(model_path, 'model.pkl'), 'rb') as f:
                cls.model = pickle.load(f)
        return cls.model

    @classmethod
    def predict(cls, x):
        clf = cls.get_model()
        return clf.predict(x)


@app.get('/ping')
def ping():
    health = ScoringService.get_model() is not None
    status_code = status.HTTP_200_OK if health else status.HTTP_404_NOT_FOUND
    return Response(status_code=status_code)


@app.post('/invocations', status_code=status.HTTP_200_OK)
def transformation(content_type: str, data: bytes):
    if content_type != 'text/csv':
        raise HTTPException(status_code=status.HTTP_415_UNSUPPORTED_MEDIA_TYPE,
                            detail='This predictor only supports CSV data')
    print('----- entre a transformation ok')
    
    data = data.decode('utf-8')
    s = io.StringIO(data)
    df = pd.read_csv(s, header=None)

    print('----- lectura data ok ok')
    
    print(f'Invoked with {df.shape[0]} records')

    predictions = ScoringService.predict(df)

    print('----- predict ok')

    out = io.StringIO()
    pd.DataFrame({'results': predictions}).to_csv(out, header=False, index=False)
    result = out.getvalue()

    print('----- conversion de results ok')

    return Response(content=result, media_type='text/csv')

Batch Transform Job配置 #

client.create_transform_job(
    TransformJobName=f'mlbatchtransform-{datetime.datetime.now().strftime("%Y%m%d%M%S")}',
    ModelName='mlbatchtransform',
    MaxConcurrentTransforms=1,
    BatchStrategy='MultiRecord',
    ModelClientConfig={
        'InvocationsTimeoutInSeconds': 100,
        'InvocationsMaxRetries': 3
    },
    MaxPayloadInMB=50,
    TransformInput={
        'DataSource': {
            'S3DataSource': {
                'S3DataType': 'S3Prefix',
                'S3Uri': 's3://<bucket>/mlbatchtransform/inference/input/inference.csv',
            }
        },
        'ContentType': 'text/csv',
        'CompressionType': 'None',
        'SplitType': 'Line',
    },
    TransformOutput={
        'S3OutputPath': 's3://<bucket>/mlbatchtransform/inference/output',
        'Accept': 'text/csv',
        'AssembleWith': 'None',
    },
    DataCaptureConfig={
        'DestinationS3Uri': 's3://<bucket>/mlbatchtransform/datacapture',
        'GenerateInferenceId': True
    },
    TransformResources={
        'InstanceType': 'ml.m5.large',
        'InstanceCount': 1,
    },
)

问题排查与解决方案 #

1. FastAPI接口参数解析错误（核心原因） #

当前/invocations接口定义的content_type: str和data: bytes参数，FastAPI默认会将其解析为查询参数或表单字段，但SageMaker是通过HTTP请求的Content-Type Header传递类型信息，并将CSV数据放在请求体中。这导致FastAPI无法找到预期的参数，返回422错误。

修改接口参数获取方式，使用Request对象直接读取Header和请求体：

from fastapi import Request

@app.post('/invocations', status_code=status.HTTP_200_OK)
async def transformation(request: Request):
    # 从Header获取Content-Type
    content_type = request.headers.get('Content-Type')
    if not content_type or content_type != 'text/csv':
        raise HTTPException(status_code=status.HTTP_415_UNSUPPORTED_MEDIA_TYPE,
                            detail='This predictor only supports CSV data')
    
    # 读取请求体数据
    data = await request.body()
    
    # 后续处理逻辑保持不变
    data = data.decode('utf-8')
    s = io.StringIO(data)
    df = pd.read_csv(s, header=None)
    
    print(f'Invoked with {df.shape[0]} records')
    predictions = ScoringService.predict(df)
    
    out = io.StringIO()
    pd.DataFrame({'results': predictions}).to_csv(out, header=False, index=False)
    result = out.getvalue()
    
    return Response(content=result, media_type='text/csv')

2. 验证服务启动脚本 #

确保serve脚本正确启动FastAPI服务，且具备可执行权限：

• serve脚本内容示例：

  #!/bin/bash
  uvicorn predictor:app --host 0.0.0.0 --port 8080
  

• 构建镜像时需添加权限设置，在Dockerfile中加入：

  RUN chmod +x /opt/program/serve
  

3. 调试请求内容 #

在接口中添加日志，打印接收到的原始数据，排查数据格式是否符合预期：

print(f"Received Content-Type: {content_type}")
print(f"Raw request body snippet: {data[:200]}...")  # 打印前200个字符

4. 确认Batch Transform配置匹配 #

• BatchStrategy='MultiRecord'会将多条CSV记录打包发送，确保pd.read_csv(header=None)能正确解析多行数据
• SplitType='Line'确保按行拆分输入数据，与你的CSV格式匹配

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