如何使用 lambda 函数将 numpy 数组发送到 sagemaker 端点

Question

如何使用输入数据类型调用 sagemaker 端点numpy.ndarray。我已经部署了一个 sagemaker 模型并尝试使用 lambda 函数来实现它。但我无法弄清楚如何去做。我收到服务器错误。

一行输入数据。总数据集有shape=(91,5,12). 下面只是一行输入数据。

array([[[0.30440741, 0.30209799, 0.33520652, 0.41558442, 0.69096432,
         0.69611016, 0.25153326, 0.98333333, 0.82352941, 0.77187154,
         0.7664042 , 0.74468085],
        [0.30894981, 0.33151662, 0.22907725, 0.46753247, 0.69437367,
         0.70410559, 0.29259044, 0.9       , 0.80882353, 0.79401993,
         0.89501312, 0.86997636],
        [0.33511896, 0.34338939, 0.24065546, 0.48051948, 0.70384005,
         0.71058715, 0.31031288, 0.86666667, 0.89705882, 0.82724252,
         0.92650919, 0.89125296],
        [0.34617355, 0.36150251, 0.23726854, 0.54545455, 0.71368726,
         0.71703244, 0.30228356, 0.85      , 0.86764706, 0.86157254,
         0.97112861, 0.94089835],
        [0.36269508, 0.35923332, 0.40285461, 0.62337662, 0.73325475,
         0.7274392 , 0.26241391, 0.85      , 0.82352941, 0.89922481,
         0.9343832 , 0.90780142]]])

我正在使用以下代码但无法调用端点

import boto3
def lambda_handler(event, context):
    # The SageMaker runtime is what allows us to invoke the endpoint that we've created.
    runtime = boto3.Session().client('sagemaker-runtime')

    endpoint = 'sagemaker-tensorflow-2019-04-22-07-16-51-717'

    print('givendata ', event['body'])
    # data = numpy.array([numpy.array(xi) for xi in event['body']])
    data = event['body']
    print('numpy array ', data)

    # Now we use the SageMaker runtime to invoke our endpoint, sending the review we were given
    response = runtime.invoke_endpoint(EndpointName = endpoint,# The name of the endpoint we created
                                       ContentType = 'application/json',                 # The data format that is expected
                                       Body = data) # The actual review

    # The response is an HTTP response whose body contains the result of our inference
    result = response['Body'].read().decode('utf-8')

    print('response', result)

    # Round the result so that our web app only gets '1' or '0' as a response.
    result = round(float(result))

    return {
        'statusCode' : 200,
        'headers' : { 'Content-Type' : 'text/plain', 'Access-Control-Allow-Origin' : '*' },
        'body' : str(result)
    }

我无法弄清楚应该用什么来代替 ContentType。因为我不知道numpy.ndarray.

Answer 1

我所拥有的以及我如何解决的插图

from sagemaker.tensorflow import TensorFlowPredictor

predictor = TensorFlowPredictor('sagemaker-tensorflow-serving-date')
data = np.array(raw_data)
response = predictor.predict(data=data)
predictions = response['predictions']
print(predictions)

我做了什么来找到答案：

• 在 sagemaker python 库中查找 predictions.py 和 content_types.py 实现以查看它使用了哪些内容类型以及它具有哪些参数。
• 首先，我认为使用了application/x-npycontent_type，因此尝试使用 predictor.py 中的序列化代码并将application/x-npyas content_type传递给 invoke_endpoint。
• 收到415（不支持的媒体类型）后，问题依旧是content_type。以下打印语句帮助我揭示了 content_type 预测器实际使用的内容 ( application/json)，因此我从 predictor.py 中获取了适当的序列化代码

from sagemaker.tensorflow import TensorFlowPredictor

predictor = TensorFlowPredictor('sagemaker-tensorflow-serving-date')
data = np.array(raw_data)
response = predictor.predict(data=data)
print(predictor.content_type)
print(predictor.accept)
predictions = response['predictions']
print(predictions)

TL; 博士

lambda 的解决方案：

import json
import boto3

ENDPOINT_NAME = 'sagemaker-tensorflow-serving-date'
config = botocore.config.Config(read_timeout=80)
runtime= boto3.client('runtime.sagemaker', config=config)
data = np.array(raw_data)
payload = json.dumps(data.tolist())
response = runtime.invoke_endpoint(EndpointName=ENDPOINT_NAME,
                                   ContentType='application/json',
                                   Body=payload)
result = json.loads(response['Body'].read().decode())
res = result['predictions']

注意： numpy 不包含在 lambda 中，因此您要么自己包含 numpy，要么使用 python 列表和 json.dump 该列表（列表）代替 data.tolist() 操作。从你的代码来看，在我看来你有 python 列表而不是 numpy 数组，所以简单的 json 转储应该可以工作。

Answer 2

如果您使用 TensorFlow 在 SageMaker 上训练和托管自定义算法，则可以将请求和响应格式序列化/反序列化为 JSON，如 TensorFlow Serving Predict API中所示。

import numpy
from sagemaker.predictor import json_serializer, json_deserializer

# define predictor
predictor = estimator.deploy(1, instance_type)

# format request
data = {'instances': numpy.asarray(np_array).astype(float).tolist()}

# set predictor request/response formats
predictor.accept = 'application/json'
predictor.content_type = 'application/json'

predictor.serializer = json_serializer
predictor.deserializer = json_deserializer

# run inference using SageMaker predict class
# https://github.com/aws/sagemaker-python-sdk/blob/master/src/sagemaker/predictor.py
predictor.predict(data)

您可以参考此处的示例笔记本来训练和托管自定义 TensorFlow 容器。