从实时端点捕获数据 - 亚马逊 SageMaker AI
如何启用数据捕获部署模型查看捕获的数据

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

从实时端点捕获数据

注意

为了防止对推理请求产生影响，数据捕获功能会在磁盘利用率较高时停止捕获请求。建议将磁盘利用率保持在 75% 以下，以确保数据捕获功能继续捕获请求。

要为实时终端节点捕获数据，必须使用 SageMaker AI 托管服务部署模型。这需要您创建 A SageMaker I 模型、定义终端节点配置并创建 HTTPS 终端节点。

无论你使用还是 SageMaker Python SDK，开启数据采集所需的步骤都是相似的。 AWS SDK for Python (Boto) 如果您使用 AWS SDK，请在CreateEndpointConfig方法中定义DataCaptureConfig字典以及必填字段，以开启数据采集。如果您使用 SageMaker Python SDK，请导入该DataCaptureConfig类并从该类初始化一个实例。然后，将此对象传递到 sagemaker.model.Model.deploy() 方法中的 DataCaptureConfig 参数。

要使用后续的代码片段，请用您自己的信息替换示例代码italicized placeholder text中的。

如何启用数据捕获

指定数据捕获配置。可以使用此配置捕获请求负载和/或响应负载。接下来的代码片段演示了如何使用 AWS SDK for Python (Boto) 和 SageMaker AI Python SDK 启用数据采集。

注意

您无需使用 Model Monitor 来捕获请求或响应负载。

AWS SDK for Python (Boto)

使用CreateEndpointConfig方法创建端点时，使用DataCaptureConfig字典配置要捕获的数据。将 EnableCapture 设置为布尔值 True。此外，还需提供以下必填参数：

  • EndpointConfigName：端点配置的名称。您在提出 CreateEndpoint 请求时将使用此名称。

  • ProductionVariants：要在此端点上托管的模型的列表。为每个模型定义字典数据类型。

  • DataCaptureConfig：字典数据类型，您可以在其中指定一个整数值，该值对应于示例 (InitialSamplingPercentage) 的初始数据百分比、要存储捕获数据的 Amazon S3 URI 以及捕获选项 (CaptureOptions) 列表。在 CaptureOptions 列表中为 CaptureMode 指定 Input 或 Output。

您可以选择通过向字典传递键值对参数来指定 SageMaker AI 应如何对捕获的数据进行编码。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 containing the captured data
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
    }
)

有关其他终端节点配置选项的更多信息，请参阅 Amazon A CreateEndpointConfig SageMaker I 服务 API 参考指南中的 API。

SageMaker Python SDK

从 sagemaker.model_monitor 模块导入 DataCaptureConfig 类。通过将 EnableCapture 设置为布尔值 True 来启用数据捕获。

（可选）为以下参数提供参数：

  • SamplingPercentage：一个整数值，对应于要采样的数据的百分比。如果您未提供采样百分比， SageMaker AI 将对默认的 20 (20%) 数据进行采样。

  • DestinationS3Uri：Amazon S3 UR SageMaker I 人工智能将用于存储捕获的数据。如果您不提供， SageMaker AI 会将捕获的数据存储在中"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 stored captured-data location
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 = ["REQUEST", "RESPONSE"],
)

部署模型

部署您的模型并创建一个启用了 DataCapture 的 HTTPS 端点。

AWS SDK for Python (Boto3)

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

完成模型和端点配置后，可使用 CreateEndpoint API 创建端点。终端节点名称在您 AWS 账户的某个 AWS 区域内必须是唯一的。

下文将使用在请求中指定的端点配置创建端点。Amazon SageMaker AI 使用终端节点来配置资源和部署模型。

# 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 AI 会为你创建一个唯一的名字：

from datetime import datetime

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

使用模型对象的内置 deploy() 方法将模型部署到实时 HTTPS 端点。在字段中提供要将此模型部署到的 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 SDK 预测器类中创建预测器对象。在以后的一个步骤中，您将使用 Predictor 类返回的对象来调用端点。提供端点的名称（前面定义为 endpoint_name），以及分别用于序列化程序和反序列化程序的序列化程序对象和反序列化程序对象。有关序列化器类型的信息，请参阅 AI SageMaker Python SDK 中的序列化器类。

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())

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

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!")

由于您在前面的步骤中启用了数据捕获，因此，请求和响应负载将与其他一些元数据一起保存到您在 DataCaptureConfig 中指定的 Amazon S3 位置。将捕获数据传输到 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 AI 的 JSOnline 格式的文件。以下是从我们使用 csv/text 数据调用的实时端点获取的响应示例：

{"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 AI 模型和捕获的数据，返回的值将有所不同：

{
  "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"
}