SageMaker批量推理数据预处理报错求助：特征数量不匹配

问题描述 #

在AWS SageMaker上使用训练完成的模型执行批量推理（Batch Transform）时，JSON格式的未预处理数据在input_fn中完成列删除、独热编码和标准化后，触发特征数不匹配错误：

ValueError: X has 37 features, but ColumnTransformer is expecting 16 features as input

实际数据包含16个特征+1个类别列，尝试两种调整方案均失败：

• 将strategy="MultiRecord"改为SingleRecord，报错变为“X has 8 features, but ColumnTransformer is expecting 16”
• 移除input_fn中的预处理步骤，因输入为非向量化数据导致推理失败

注：相同代码在Serverless实时推理场景下可正常运行。

错误根源 #

核心问题在于推理阶段重新创建了全新的ColumnTransformer并调用fit_transform：

1. 训练时的预处理逻辑（列删除、独热编码、标准化）已包含在保存的pipeline_model.joblib中，且已通过训练数据拟合出固定参数（如StandardScaler的均值/方差、OneHotEncoder的类别集合）
2. 推理时在input_fn中新建的ColumnTransformer未复用训练参数，调用fit_transform会用当前批量数据重新拟合，导致输出特征维度与训练时模型期望的维度不一致
3. 批量推理的MultiRecord/SingleRecord数据拆分策略，进一步放大了重新拟合带来的特征数差异

修正方案 #

直接复用训练好的Pipeline中的预处理逻辑，无需在input_fn中重复实现预处理步骤，仅保留数据格式转换和类别列删除：

修改后的serve.py #

%%writefile serve.py

import os
import joblib
import pandas as pd

def model_fn(model_dir):
    """加载并返回训练好的完整Pipeline模型"""
    model_file_name = "pipeline_model.joblib"
    pipeline_model = joblib.load(os.path.join(model_dir, model_file_name))
    
    return pipeline_model
      
def input_fn(request_body, request_content_type):
    """处理输入JSON数据：转换为DataFrame并移除类别列"""
    if request_content_type == "application/json":
        input_object = pd.read_json(request_body, lines=True)
        input_object.drop(["y"], axis=1, inplace=True)
        return input_object
    else:
        raise ValueError("仅支持application/json格式输入！")  

def predict_fn(input_object, pipeline_model):
    """利用完整Pipeline完成预处理+预测"""
    predictions = pipeline_model.predict(input_object)
    pred_probs = pipeline_model.predict_proba(input_object)
    
    prediction_object = pd.DataFrame(
        {
            "prediction": predictions.tolist(),
            "pred_prob_class0": pred_probs[:, 0].tolist(),
            "pred_prob_class1": pred_probs[:, 1].tolist()
        }
    )
    
    return prediction_object

def output_fn(prediction_object, request_content_type):
    """将预测结果转换为JSON格式输出"""
    return_object = prediction_object.to_json(orient="records", lines=True)
    
    return return_object

关键修改说明 #

• 删除input_fn中手动创建OneHotEncoder、StandardScaler、ColumnTransformer的代码，避免重复拟合预处理参数
• 仅保留数据格式转换和类别列删除操作，直接将原始特征DataFrame传入predict_fn
• 依赖训练好的pipeline_model自动执行预处理逻辑，确保推理阶段与训练阶段的预处理行为完全一致

批量推理配置无需调整 #

原有的批量Transformer配置可直接复用：

batch_transformer = model.transformer(instance_count=1, 
                                      instance_type="ml.m5.xlarge",
                                      strategy="MultiRecord",
                                      accept="application/json",
                                      assemble_with="Line", 
                                      output_path=output_path)

额外验证建议 #

1. 确认训练时的Pipeline确实包含完整的预处理流程（列删除、独热编码、标准化），确保pipeline_model.joblib保存了预处理+模型的完整链路
2. 本地测试：下载模型文件，加载后用单条测试数据验证pipeline_model.predict(test_df)是否能正常输出结果
3. 小批量验证：先上传少量测试数据到S3，确认批量推理任务可正常执行后再处理全量数据

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