检查预测结果 - Amazon SageMaker
Amazon Web Services 文档中描述的 Amazon Web Services 服务或功能可能因区域而异。要查看适用于中国区域的差异,请参阅 中国的 Amazon Web Services 服务入门 (PDF)

检查预测结果

您可以使用多种方法来检查异步端点的预测结果。这些选项包括:

  1. Amazon SNS 主题。

  2. 在 Amazon S3 存储桶中查看输出。

Amazon SNS 主题

Amazon SNS 是面向消息收发应用程序的通知服务,多个订阅用户可以选择通过多种传输协议(包括 HTTP、Amazon SQS 和电子邮件)请求和接收注重时效消息的“推送”通知。当您使用 CreateEndpointConfig 创建端点并指定了 Amazon SNS 主题时,Amazon SageMaker 异步推理会发布通知。

注意

要接收 Amazon SNS 通知,您的 IAM 角色必须具有 sns:Publish 权限。有关使用异步推理必须满足的要求的信息,请参阅先决条件

要使用 Amazon SNS 检查异步端点的预测结果,您首先需要创建一个主题,订阅并确认订阅该主题,然后记下该主题的 Amazon 资源名称 (ARN)。有关如何创建、订阅和查找 Amazon SNS 主题的 Amazon ARN 的详细信息,请参阅配置 Amazon SNS

使用 CreateEndpointConfig 创建端点配置时,请在 AsyncInferenceConfig 字段中提供 Amazon SNS 主题 ARN。您可以同时指定 Amazon SNS ErrorTopicSuccessTopic

import boto3 sagemaker_client = boto3.client('sagemaker', region_name=<aws_region>) sagemaker_client.create_endpoint_config( EndpointConfigName=<endpoint_config_name>, # You specify this name in a CreateEndpoint request. # List of ProductionVariant objects, one for each model that you want to host at this endpoint. ProductionVariants=[ { "VariantName": "variant1", # The name of the production variant. "ModelName": "model_name", "InstanceType": "ml.m5.xlarge", # Specify the compute instance type. "InitialInstanceCount": 1 # Number of instances to launch initially. } ], AsyncInferenceConfig={ "OutputConfig": { # Location to upload response outputs when no location is provided in the request. "S3OutputPath": "s3://<bucket>/<output_directory>" "NotificationConfig": { "SuccessTopic": "arn:aws:sns:aws-region:account-id:topic-name", "ErrorTopic": "arn:aws:sns:aws-region:account-id:topic-name", } } } )

创建端点并进行调用后,您将收到来自 Amazon SNS 主题的通知。例如,如果您进行订阅以从主题接收电子邮件通知,那么每次调用端点时,您都会收到一封电子邮件通知。以下示例显示成功调用电子邮件通知的 JSON 内容。

{ "awsRegion":"us-east-1", "eventTime":"2022-01-25T22:46:00.608Z", "receivedTime":"2022-01-25T22:46:00.455Z", "invocationStatus":"Completed", "requestParameters":{ "contentType":"text/csv", "endpointName":"<example-endpoint>", "inputLocation":"s3://<bucket>/<input-directory>/input-data.csv" }, "responseParameters":{ "contentType":"text/csv; charset=utf-8", "outputLocation":"s3://<bucket>/<output_directory>/prediction.out" }, "inferenceId":"11111111-2222-3333-4444-555555555555", "eventVersion":"1.0", "eventSource":"aws:sagemaker", "eventName":"InferenceResult" }

检查您的 S3 存储桶

当您使用 InvokeEndpointAsync 调用端点时,它会返回一个响应对象。您可以使用响应对象获取用于存储输出的 Amazon S3 URI。有了输出位置,您就可以使用 SageMaker Python SDK SageMaker 会话类,以编程方式检查输出。

以下命令将 InvokeEndpointAsync 的输出字典存储为以变量命名的响应。使用响应变量,您可以随后获取 Amazon S3 输出 URI,并将其存储为名为 output_location 的字符串变量。

import uuid import boto3 sagemaker_runtime = boto3.client("sagemaker-runtime", region_name=<aws_region>) # Specify the S3 URI of the input. Here, a single SVM sample input_location = "s3://bucket-name/test_point_0.libsvm" response = sagemaker_runtime.invoke_endpoint_async( EndpointName='<endpoint-name>', InputLocation=input_location, InferenceId=str(uuid.uuid4()), ContentType="text/libsvm" #Specify the content type of your data ) output_location = response['OutputLocation'] print(f"OutputLocation: {output_location}")

有关支持的内容类型的信息,请参阅用于推理的常见数据格式

有了 Amazon S3 的输出位置,随后您就可以使用 SageMaker Python SDK SageMaker 会话类来读入 Amazon S3 文件。下面的代码示例演示了如何创建函数 (get_ouput),该函数会反复尝试从 Amazon S3 输出位置读取文件:

import sagemaker import urllib, time from botocore.exceptions import ClientError sagemaker_session = sagemaker.session.Session() def get_output(output_location): output_url = urllib.parse.urlparse(output_location) bucket = output_url.netloc key = output_url.path[1:] while True: try: return sagemaker_session.read_s3_file( bucket=output_url.netloc, key_prefix=output_url.path[1:]) except ClientError as e: if e.response['Error']['Code'] == 'NoSuchKey': print("waiting for output...") time.sleep(2) continue raise output = get_output(output_location) print(f"Output: {output}")