基于SageMaker集成KNN模型预测时遇Record解析错误求助

基于SageMaker集成建模示例运行多模型集成预测，已成功在原有XGBoost、线性模型基础上添加KNN模型，但预测时触发CloudWatch报错：

Customer Error: Record could not be parsed to valid data point. Please make sure the data is valid and try again. (caused by KeyError)

集成模型与端点构建代码 #

# build a model

sm_client = boto3.client('sagemaker')

model_name = "MODEL-for-ensemble-modelling-" + strftime("%Y-%m-%d-%H-%M-%S", gmtime())
xgboost_hosting_container = {
    "Image": xgb_container,
    "ContainerHostname": "xgboost",
    "ModelDataUrl": sm.describe_training_job(TrainingJobName=xgboost_job_name)
    ["ModelArtifacts"]["S3ModelArtifacts"],
}

linear_hosting_container = {
    "Image": container,
    "ContainerHostname": "linear",
    "ModelDataUrl": sm.describe_training_job(TrainingJobName=linear_job_name)["ModelArtifacts"][
        "S3ModelArtifacts"
    ],
}

knn_hosting_container = {
    "Image": knn_container,
    "ContainerHostname": "knn",
    "ModelDataUrl": sm.describe_training_job(TrainingJobName=knn_job_name)["ModelArtifacts"][
        "S3ModelArtifacts"
    ],
}

#inferenceExecutionConfig = {"Mode": "Direct"}

model = sm_client.create_model( ModelName=model_name,
                                  #InferenceExecutionConfig=inferenceExecutionConfig,
                                  ExecutionRoleArn=role,
                                  Containers=[xgboost_hosting_container, linear_hosting_container, knn_hosting_container],)

print('Model created: '+ model['ModelArn'])

#Specify data configuration setting

s3_capture_upload_path = 's3://{}/{}/monitoring/datacapture'.format(rawbucket, prefix)

data_capture_configuration = {
    "EnableCapture": True,
    "InitialSamplingPercentage": 100,
    "DestinationS3Uri": s3_capture_upload_path,
    "CaptureOptions": [
        { "CaptureMode": "Output" },
        { "CaptureMode": "Input" }
    ],
    "CaptureContentTypeHeader": {
       "CsvContentTypes": ["text/csv"],
       "JsonContentTypes": ["application/json"]}}

#Create endpoint configuration

def create_endpoint_config(model_config, data_capture_config): 
    return sm_client.create_endpoint_config(
                                                EndpointConfigName=model_config,
                                                ProductionVariants=[
                                                        {
                                                            'VariantName': 'AllTraffic',
                                                            'ModelName': model_name,
                                                            'InitialInstanceCount': 1,
                                                            'InstanceType': 'ml.m4.xlarge',
                                                            'InitialVariantWeight': 1.0,
                                                },
                                                    
                                                    ],
                                                DataCaptureConfig=data_capture_config
                                                )

endpoint_config = create_endpoint_config(model_name, data_capture_configuration)

print('Endpoint configuration created: '+ endpoint_config['EndpointConfigArn'])

#Create endpoint

def create_endpoint(endpoint_name, config_name):
    return sm_client.create_endpoint(
                                    EndpointName=model_name,
                                    EndpointConfigName=model_name
                                )
endpoint = create_endpoint(model_name, endpoint_config)

print('Endpoint created: '+ endpoint['EndpointArn'])    

预测代码 #

datas = pd.read_csv('test_data.csv',header = None)

datas = datas.drop(columns = 0)

datas = datas.sample(n=87)

#Sending inference requests to the multimodel endpoint

from sagemaker.predictor import Predictor
from sagemaker.serializers import CSVSerializer
from sagemaker.deserializers import JSONDeserializer

predictor = Predictor(endpoint_name=model_name,
                      serializer=CSVSerializer(),
                      deserializer=JSONDeserializer())
response = predictor.predict(datas.values)

print('done!')

报错回溯信息 #

ModelError                                Traceback (most recent call last)
<ipython-input-77-f8fe7afa50b9> in <module>
      8                       serializer=CSVSerializer(),
      9                       deserializer=JSONDeserializer())
---> 10 response = predictor.predict(datas.values)
     11 
     12 print('done!')

/opt/conda/lib/python3.7/site-packages/sagemaker/predictor.py in predict(self, data, initial_args, target_model, target_variant, inference_id)
    159             data, initial_args, target_model, target_variant, inference_id
    160         )
--> 161         response = self.sagemaker_session.sagemaker_runtime_client.invoke_endpoint(**request_args)
    162         return self._handle_response(response)
    163 

/opt/conda/lib/python3.7/site-packages/botocore/client.py in _api_call(self, *args, **kwargs)
    528                 )
    529             # The "self" in this scope is referring to the BaseClient.
--> 530             return self._make_api_call(operation_name, kwargs)
    531 
    532         _api_call.__name__ = str(py_operation_name)

/opt/conda/lib/python3.7/site-packages/botocore/client.py in _make_api_call(self, operation_name, api_params)
    958             error_code = parsed_response.get("Error", {}).get("Code")
    959             error_class = self.exceptions.from_code(error_code)
--> 960             raise error_class(parsed_response, operation_name)
    961         else:
    962             return parsed_response

ModelError: An error occurred (ModelError) when calling the InvokeEndpoint operation: Received client error (400) from knn with message "unable to evaluate payload provided".

CloudWatch日志详情 #

Customer Error: Record could not be parsed to valid data point. Please make sure the data is valid and try again. (caused by KeyError)
Caused by: 'instances'

Traceback (most recent call last):
  File "/opt/amazon/lib/python3.7/site-packages/ai_algorithms_sdk/io/serve_helpers.py", line 78, in _iterator_points_dense_rank_2
    points = get_points(payload)
  File "/opt/amazon/lib/python3.7/site-packages/ai_algorithms_sdk/io/serve_helpers.py", line 93, in _get_points
    return [_load_json_instance(instance) for instance in raw_data["instances"]]

KeyError: 'instances'

During handling of the above exception, another exception occurred:

Traceback (most recent call last):
  File "/opt/amazon/lib/python3.7/site-packages/ai_algorithms_sdk/serve.py", line 722, in get_invocation_response
    data_iter = get_data_iterator(payload, **content_parameters)
  File "/opt/amazon/lib/python3.7/site-packages/ai_algorithms_sdk/io/serve_helpers.py", line 94, in iterator_json_dense_rank_2
    return _iterator_points_dense_rank_2(payload, _get_points)
  File "/opt/amazon/lib/python3.7/site-packages/ai_algorithms_sdk/io/serve_helpers.py", line 80, in _iterator_points_dense_rank_2
    raise NonParsableRecordCustomerError(caused_by=e)

ai_algorithms_sdk.io.serve_helpers.NonParsableRecordCustomerError: Record could not be parsed to valid data point. Please make sure the data is valid and try again. (caused by KeyError)

问题原因 #

SageMaker的KNN内置容器默认期望JSON格式输入，且结构需包含instances字段，但当前预测代码使用CSVSerializer传递CSV格式数据，导致KNN模型无法解析输入，触发KeyError: 'instances'。而原有XGBoost和线性模型支持CSV格式，多模型集成端点无法同时兼容两种输入格式，除非配置Direct模式或统一输入格式。

解决方法 #

方法1：启用Direct模式，指定目标模型调用

1. 创建模型时启用Direct模式：

inferenceExecutionConfig = {"Mode": "Direct"}

model = sm_client.create_model( 
    ModelName=model_name,
    InferenceExecutionConfig=inferenceExecutionConfig,
    ExecutionRoleArn=role,
    Containers=[xgboost_hosting_container, linear_hosting_container, knn_hosting_container],
)

2. 预测时分别调用不同模型，使用对应序列化器：

# 调用XGBoost/线性模型（CSV格式）
predictor_csv = Predictor(endpoint_name=model_name,
                          serializer=CSVSerializer(),
                          deserializer=JSONDeserializer())
response_xgb = predictor_csv.predict(datas.values, target_model="xgboost")

# 调用KNN模型（JSON格式）
from sagemaker.serializers import JSONSerializer
predictor_knn = Predictor(endpoint_name=model_name,
                          serializer=JSONSerializer(),
                          deserializer=JSONDeserializer())
# 构造KNN要求的输入结构
knn_input = {"instances": datas.values.tolist()}
response_knn = predictor_knn.predict(knn_input, target_model="knn")

方法2：统一使用JSON格式输入

修改所有模型的输入格式为JSON，确保兼容KNN：

from sagemaker.serializers import JSONSerializer

predictor = Predictor(endpoint_name=model_name,
                      serializer=JSONSerializer(),
                      deserializer=JSONDeserializer())
# 构造兼容所有模型的JSON输入
input_data = {"instances": datas.values.tolist()}
response = predictor.predict(input_data)

注：SageMaker的XGBoost和线性学习器容器均支持该JSON结构输入。

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