捕获数据 - Amazon SageMaker
Amazon Web Services 文档中描述的 Amazon Web Services 服务或功能可能因区域而异。要查看适用于中国区域的差异,请参阅中国的 Amazon Web Services 服务入门

本文属于机器翻译版本。若本译文内容与英语原文存在差异,则一律以英文原文为准。

捕获数据

将输入记录到终端节点和推理输出 SageMaker 到 Amazon S3 的实时终端节点,您可以启用名为数据捕获. 它通常用于记录可用于训练、调试和监控的信息。亚马逊 SageMaker Model Monitor 会自动解析这些捕获的数据,并将此数据中的指标与您为模型创建的基线进行比较。有关模型监控器的更多信息,请监控模型的数据和模型质量、偏见和可解释性.

注意

为了防止对推理请求的影响,Data Capture 停止捕获高磁盘使用率的请求。建议将磁盘利用率保持在 75% 以下,以确保数据捕获继续捕获请求。

要捕获数据,必须使用 SageMaker 托管服务。这要求你创建 SageMaker 模型、定义终端节点配置并创建 HTTPS 终端节点。

启用数据捕获所需的步骤都类似,无论您是否使用Amazon SDK for Python (Boto)或者 SageMaker Python 开发工具包。如果您将AmazonSDK,定义DataCaptureConfig字典以及必填字段在CreateEndpointConfig启用数据捕获的方法。如果您将 SageMaker Python SDK,导入DataCaptureConfigClass 并从此类中初始化实例。然后,将此对象传递到DataCaptureConfig中的参数sagemaker.model.Model.deploy()方法。

要使用继续执行的代码片段,请替换斜体占位符文本在示例代码中,包含您自己的信息。

如何启用数据捕获

指定数据捕获配置。可以使用此配置捕获请求负载和/或响应负载。继续的代码片段演示了如何使用Amazon SDK for Python (Boto)和 SageMaker Python 开发工具包。

注意

您不需要使用模型监视器来捕获请求或响应有效载荷。

Amazon SDK for Python (Boto)

使用配置要捕获的数据DataCaptureConfig使用创建终端节点时使用CreateEndpointConfig方法。SetEnableCapture到布尔值 True。此外,还提供以下必备参数:

  • EndpointConfigName:终端节点配置的名称。当你制作CreateEndpoint请求.

  • ProductionVariants:您想在此终端节点上托管的型号列表。为每个模型定义字典数据类型。

  • DataCaptureConfig:字典数据类型,您可以在其中指定一个整数值,该值与要采样的数据的初始百分比相对应 (InitialSamplingPercentage)、您希望存储捕获数据的 Amazon S3 URI,以及捕获选项(CaptureOptions列表。指定Input要么Output为了CaptureMode中的CaptureOptions列表。

您可以选择指定方式 SageMaker 应该通过将键值对参数传递给捕获的数据进行编码CaptureContentTypeHeader词典。

# Create an endpoint config name. endpoint_config_name = '<endpoint-config-name>' # The name of the production variant. variant_name = '<name-of-production-variant>' # The name of the model that you want to host. # This is the name that you specified when creating the model. model_name = '<The_name_of_your_model>' instance_type = '<instance-type>' #instance_type='ml.m5.xlarge' # Example # Number of instances to launch initially. initial_instance_count = <integer> # Sampling percentage. Choose an integer value between 0 and 100 initial_sampling_percentage = <integer> # The S3 URI of where to store captured data in S3 s3_capture_upload_path = 's3://<bucket-name>/<data_capture_s3_key>' # Specify either Input, Output, or both capture_modes = [ "Input", "Output" ] #capture_mode = [ "Input"] # Example - If you want to capture input only endpoint_config_response = sagemaker_client.create_endpoint_config( EndpointConfigName=endpoint_config_name, # List of ProductionVariant objects, one for each model that you want to host at this endpoint. ProductionVariants=[ { "VariantName": variant_name, "ModelName": model_name, "InstanceType": instance_type, # Specify the compute instance type. "InitialInstanceCount": initial_instance_count # Number of instances to launch initially. } ], DataCaptureConfig= { 'EnableCapture': True, # Whether data should be captured or not. 'InitialSamplingPercentage' : initial_sampling_percentage, 'DestinationS3Uri': s3_capture_upload_path, 'CaptureOptions': [{"CaptureMode" : capture_mode} for capture_mode in capture_modes] # Example - Use list comprehension to capture both Input and Output } )

有关其他终端节点配置选项的更多信息,请参阅CreateEndpointConfig中的 API亚马逊 SageMaker 服务 API 参考指南.

SageMaker Python SDK

导入DataCaptureConfig来自sagemaker.model_监控器模块。通过设置启用数据捕获EnableCapture布尔值True.

可选为以下参数提供参数:

  • SamplingPercentage:一个整数值,对应于要采样的数据的百分比。如果您没有提供抽样百分比, SageMaker 将采样默认值为 20 (20%) 个数据。

  • DestinationS3Uri:Amazon S3 URI SageMaker 将用于存储捕获的数据。如果你不提供一个, SageMaker 将在中存储捕获的数据"s3://<default-session-bucket>/ model-monitor/data-capture".

from sagemaker.model_monitor import DataCaptureConfig # Set to True to enable data capture enable_capture = True # Optional - Sampling percentage. Choose an integer value between 0 and 100 sampling_percentage = <int> # sampling_percentage = 30 # Example 30% # Optional - The S3 URI of where to store captured data in S3 s3_capture_upload_path = 's3://<bucket-name>/<data_capture_s3_key>' # Specify either Input, Output or both. capture_modes = ['REQUEST','RESPONSE'] # In this example, we specify both # capture_mode = ['REQUEST'] # Example - If you want to only capture input. # Configuration object passed in when deploying Models to SM endpoints data_capture_config = DataCaptureConfig( enable_capture = enable_capture, sampling_percentage = sampling_percentage, # Optional destination_s3_uri = s3_capture_upload_path, # Optional capture_options = [{"CaptureMode": capture_mode} for capture_mode in capture_modes] )

部署模型

使用部署模型并创建 HTTPS 终端节点DataCapture已启用.

Amazon SDK for Python (Boto3)

向 SageMaker 提供端点配置。该服务会启动 ML 计算实例,并按照配置中的规定部署一个或多个模型。

拥有模型和终端节点配置后,请使用CreateEndpoint用于创建终端节点的 API。在中,终端节点名称必须唯一。Amazon您的区域Amazonaccount.

以下内容使用在请求中指定的终端节点配置创建终端节点。亚马逊 SageMaker 使用终端节点来配置资源和部署模型。

# The name of the endpoint. The name must be unique within an AWS Region in your AWS account. endpoint_name = '<endpoint-name>' # The name of the endpoint configuration associated with this endpoint. endpoint_config_name='<endpoint-config-name>' create_endpoint_response = sagemaker_client.create_endpoint( EndpointName=endpoint_name, EndpointConfigName=endpoint_config_name)

有关更多信息,请参阅 CreateEndpoint API。

SageMaker Python SDK

为终端节点定义名称。此为可选步骤。如果你不提供一个, SageMaker 将为你创建一个唯一的名字:

from datetime import datetime endpoint_name = f"DEMO-{datetime.utcnow():%Y-%m-%d-%H%M}" print("EndpointName =", endpoint_name)

使用模型对象的内置将模型部署到实时的 HTTPS 终端节点deploy()方法。提供要将此模型部署到的 Amazon EC2 实例类型的名称instance_type此字段以及要在上运行终端节点的终端节点的实例的初始数量以及initial_instance_count字段:

initial_instance_count=<integer> # initial_instance_count=1 # Example instance_type='<instance-type>' # instance_type='ml.m4.xlarge' # Example # Uncomment if you did not define this variable in the previous step #data_capture_config = <name-of-data-capture-configuration> model.deploy( initial_instance_count=initial_instance_count, instance_type=instance_type, endpoint_name=endpoint_name, data_capture_config=data_capture_config )

查看捕获的数据

从中创建预测器对象 SageMaker Python 开发工具包预测器类。您将使用返回的对象Predictor在 future 的步骤中调用终端节点的类。提供终端节点的名称(之前定义为endpoint_name),以及分别用于序列化器和反序列化器的序列化器和反序列化器对象。有关序列化器类型的信息,请参阅序列化器中的课SageMaker Python 开发工具包.

from sagemaker.predictor import Predictor from sagemaker.serializers import <Serializer> from sagemaker.deserializers import <Deserializers> predictor = Predictor(endpoint_name=endpoint_name, serializer = <Serializer_Class>, deserializer = <Deserializer_Class>) # Example #from sagemaker.predictor import Predictor #from sagemaker.serializers import CSVSerializer #from sagemaker.deserializers import JSONDeserializer #predictor = Predictor(endpoint_name=endpoint_name, # serializer=CSVSerializer(), # deserializer=JSONDeserializer())

在接下来的代码示例场景中,我们使用我们在本地存储在名为的 CSV 文件中的示例验证数据调用终端节点。validation_with_predictions. 我们的示例验证集包含每个输入的标签。

WITH 语句的前几行首先打开验证集 CSV 文件,然后用逗号字符拆分文件中的每一行",",然后将两个返回的对象存储到标签和 input_cols 变量中。对于每一行,输入 (input_cols) 被传递给预测变量的 (predictor) 对象内置方法Predictor.predict().

假设模型返回概率。整数值为 0 和 1.0 之间的概率范围。如果模型返回的概率大于 80% (0.8),我们将预测分配一个整数值标签 1。否则,我们为预测分配一个整数值标签 0。

from time import sleep validate_dataset = "validation_with_predictions.csv" # Cut off threshold of 80% cutoff = 0.8 limit = 200 # Need at least 200 samples to compute standard deviations i = 0 with open(f"test_data/{validate_dataset}", "w") as validation_file: validation_file.write("probability,prediction,label\n") # CSV header with open("test_data/validation.csv", "r") as f: for row in f: (label, input_cols) = row.split(",", 1) probability = float(predictor.predict(input_cols)) prediction = "1" if probability > cutoff else "0" baseline_file.write(f"{probability},{prediction},{label}\n") i += 1 if i > limit: break print(".", end="", flush=True) sleep(0.5) print() print("Done!")

由于您在前面的步骤中启用了数据捕获,因此,请求和响应负载将与其他一些元数据一起保存到您在中指定的 Amazon S3 位置。DataCaptureConfig. 向 Amazon S3 交付捕获数据可能需要几分钟时间。

通过列出 Amazon S3 中存储的数据捕获文件来查看捕获的数据。Amazon S3 路径的格式为:s3:///{endpoint-name}/{variant-name}/yyyy/mm/dd/hh/filename.jsonl.

预计会看到根据调用发生时间组织的来自各个时段的不同文件。运行以下命令以打印出单个捕获文件的内容:

print("\n".join(capture_file[-3:-1]))

这将返回 SageMaker 特定的 JSLINE 格式化的文件。以下是从我们使用调用的实时终端节点中获取的响应示例csv/textDATA:

{"captureData":{"endpointInput":{"observedContentType":"text/csv","mode":"INPUT", "data":"69,0,153.7,109,194.0,105,256.1,114,14.1,6,1,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,1,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,1,0,1,0,1,0\n", "encoding":"CSV"},"endpointOutput":{"observedContentType":"text/csv; charset=utf-8","mode":"OUTPUT","data":"0.0254181120544672","encoding":"CSV"}}, "eventMetadata":{"eventId":"aaaaaaaa-bbbb-cccc-dddd-eeeeeeeeeeee","inferenceTime":"2022-02-14T17:25:49Z"},"eventVersion":"0"} {"captureData":{"endpointInput":{"observedContentType":"text/csv","mode":"INPUT", "data":"94,23,197.1,125,214.5,136,282.2,103,9.5,5,4,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,1,0,0,0,1,0,1,0,1\n", "encoding":"CSV"},"endpointOutput":{"observedContentType":"text/csv; charset=utf-8","mode":"OUTPUT","data":"0.07675473392009735","encoding":"CSV"}}, "eventMetadata":{"eventId":"aaaaaaaa-bbbb-cccc-dddd-eeeeeeeeeeee","inferenceTime":"2022-02-14T17:25:49Z"},"eventVersion":"0"}

在前面的示例中,capture_file对象是列表类型。对列表的第一个元素进行索引以查看单个推理请求。

# The capture_file object is a list. Index the first element to view a single inference request print(json.dumps(json.loads(capture_file[0]), indent=2))

此操作将返回与以下内容类似的响应。返回的值将因您的终端节点配置而异, SageMaker 模型和捕获的数据:

{ "captureData": { "endpointInput": { "observedContentType": "text/csv", # data MIME type "mode": "INPUT", "data": "50,0,188.9,94,203.9,104,151.8,124,11.6,8,3,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,1,0,0,0,0,0,0,0,0,0,1,1,0,1,0\n", "encoding": "CSV" }, "endpointOutput": { "observedContentType": "text/csv; charset=character-encoding", "mode": "OUTPUT", "data": "0.023190177977085114", "encoding": "CSV" } }, "eventMetadata": { "eventId": "aaaaaaaa-bbbb-cccc-dddd-eeeeeeeeeeee", "inferenceTime": "2022-02-14T17:25:06Z" }, "eventVersion": "0" }