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

Question

如何使用输入数据类型numpy.ndarray调用sagemaker端点。我已经部署了一个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的情况下我不知道MIME类型。

Answer 1

如果您使用TensorFlow在SageMaker上培训和托管自定义算法，则可以将请求和响应格式序列化/反序列化为JSON，如TensorFlow服务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)

您可以在此处参考示例notebook来训练和托管自定义TensorFlow容器。