numpy - 如何使用 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 的 MIME 类型.

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-npy content_type 被使用，因此尝试使用来自 predictor.py 的序列化代码并传递 application/x-npy作为 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 转储应该可以工作。