计划监控作业

Amazon SageMaker Model Monitor 使您能够按持续时间表监控从实时终端节点收集的数据，或者按特定时间表监控从批量转换任务中收集的数据。可以使用 CreateMonitoringSchedule API 创建具有预定义的周期性间隔的监控计划。例如，每 x 小时（x 的范围是 1 到 23）。

通过监控计划， SageMaker 可以按指定频率启动处理任务，以分析在给定时段内收集的数据。 SageMaker 提供用于对表格数据集进行分析的预构建容器。在处理作业中， SageMaker 将当前分析的数据集与基线统计数据、提供的约束条件进行比较，然后生成违规报告。此外，还会为每个要分析的特征发布 CloudWatch 指标。或者，您可以选择自带容器，如构建您自己的容器主题中所述。

您可以为实时终端节点或批量转换作业创建模型监控计划。使用基准资源（约束条件和统计数据）与实时流量或批处理作业输入进行比较。在以下示例中，用于训练模型的训练数据集已上传到 Amazon S3。如果您已经在 Amazon S3 中安装了它，则可以直接指向它。

# copy over the training dataset to Amazon S3 (if you already have it in Amazon S3, you could reuse it)
baseline_prefix = prefix + '/baselining'
baseline_data_prefix = baseline_prefix + '/data'
baseline_results_prefix = baseline_prefix + '/results'

baseline_data_uri = 's3://{}/{}'.format(bucket,baseline_data_prefix)
baseline_results_uri = 's3://{}/{}'.format(bucket, baseline_results_prefix)
print('Baseline data uri: {}'.format(baseline_data_uri))
print('Baseline results uri: {}'.format(baseline_results_uri))

training_data_file = open("test_data/training-dataset-with-header.csv", 'rb')
s3_key = os.path.join(baseline_prefix, 'data', 'training-dataset-with-header.csv')
boto3.Session().resource('s3').Bucket(bucket).Object(s3_key).upload_fileobj(training_data_file)

如果您正在为实时端点安排模型监控，请使用基线限制和统计数据与实时流量进行比较。以下代码片段显示了您用于为实时终端节点调度模型监视器的一般格式。

from sagemaker.model_monitor import CronExpressionGenerator
from time import gmtime, strftime

mon_schedule_name = 'DEMO-xgb-churn-pred-model-monitor-schedule-' + strftime("%Y-%m-%d-%H-%M-%S", gmtime())
my_default_monitor.create_monitoring_schedule(
    monitor_schedule_name=mon_schedule_name,
    endpoint_input=EndpointInput(
        endpoint_name=endpoint_name,
        destination="/opt/ml/processing/input/endpoint"
    ),
    post_analytics_processor_script=s3_code_postprocessor_uri,
    output_s3_uri=s3_report_path,
    statistics=my_default_monitor.baseline_statistics(),
    constraints=my_default_monitor.suggested_constraints(),
    schedule_cron_expression=CronExpressionGenerator.hourly(),
    enable_cloudwatch_metrics=True,
)

以下代码片段显示了用于为批量转换作业调度模型监视器的一般格式。

from sagemaker.model_monitor import (
    CronExpressionGenerator,
    BatchTransformInput, 
    MonitoringDatasetFormat, 
)
from time import gmtime, strftime

mon_schedule_name = 'DEMO-xgb-churn-pred-model-monitor-schedule-' + strftime("%Y-%m-%d-%H-%M-%S", gmtime())
my_default_monitor.create_monitoring_schedule(
    monitor_schedule_name=mon_schedule_name,
    batch_transform_input=BatchTransformInput(
        destination="opt/ml/processing/input",
        data_captured_destination_s3_uri=s3_capture_upload_path,
        dataset_format=MonitoringDatasetFormat.csv(header=False),
    )
    post_analytics_processor_script=s3_code_postprocessor_uri,
    output_s3_uri=s3_report_path,
    statistics=my_default_monitor.baseline_statistics(),
    constraints=my_default_monitor.suggested_constraints(),
    schedule_cron_expression=CronExpressionGenerator.hourly(),
    enable_cloudwatch_metrics=True,
)

描述和检查计划：在描述计划后，注意由 ListMonitoringSchedules API 返回的 MonitoringScheduleSummary 中的 MonitoringScheduleStatus 会更改为 Scheduled。

desc_schedule_result = my_default_monitor.describe_schedule()
print('Schedule status: {}'.format(desc_schedule_result['MonitoringScheduleStatus']))