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

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

从实时端点捕获数据

注意

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

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

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

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

如何启用数据捕获

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

注意

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

AWS SDK for Python (Boto)

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

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

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

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

您可以选择通过向字典传递键值对参数来指定 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 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
    }
)

有关其他终端节点配置选项的更多信息，请参阅《亚马逊 SageMaker 服务API参考指南》CreateEndpointConfigAPI中的。

SageMaker Python SDK

从 sagemaker.model_monitor 模块导入 DataCaptureConfig 类。通过将 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 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。该服务会启动机器学习计算实例，并按照配置中的规定部署一个或多个模型。

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

下文将使用在请求中指定的端点配置创建端点。Amazon 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)

有关更多信息，请参阅CreateEndpointAPI。

SageMaker Python SDK

为端点定义名称。此为可选步骤。如果您不提供一个名称， SageMaker 将为您创建一个唯一的名称：

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），以及分别用于序列化程序和反序列化程序的序列化程序对象和反序列化程序对象。有关序列化器类型的信息，请参阅 Pyth on 中的序列化器类。SageMaker 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())

在接下来的代码示例场景中，我们使用本地存储在名为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!")

由于您在前面的步骤中启用了数据捕获，因此，请求和响应负载将与其他一些元数据一起保存到您在 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 JSON特定的行格式文件。以下是从我们使用 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 模型和捕获的数据，返回的值将有所不同：

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