Planification des tâches de surveillance de la qualité des données

Après avoir créé votre base de référence, vous pouvez appeler la méthode create_monitoring_schedule() de votre instance de classe DefaultModelMonitor pour planifier une surveillance horaire de la qualité des données. Les sections suivantes expliquent comment créer une surveillance de la qualité des données pour un modèle déployé sur un point de terminaison en temps réel ainsi que pour une tâche de transformation par lots.

Important

Vous pouvez spécifier une entrée de transformation par lots ou une entrée de point de terminaison, mais pas les deux, lorsque vous créez votre planification de surveillance.

Surveillance de la qualité des données pour les modèles déployés sur des points de terminaison en temps réel

Pour planifier une surveillance de la qualité des données pour un point de terminaison en temps réel, transmettez votre instance EndpointInput à l'argument endpoint_input de votre instance DefaultModelMonitor, comme indiqué dans l'exemple de code suivant :

from sagemaker.model_monitor import CronExpressionGenerator
                
data_quality_model_monitor = DefaultModelMonitor(
   role=sagemaker.get_execution_role(),
   ...
)

schedule = data_quality_model_monitor.create_monitoring_schedule(
   monitor_schedule_name=schedule_name,
   post_analytics_processor_script=s3_code_postprocessor_uri,
   output_s3_uri=s3_report_path,
   schedule_cron_expression=CronExpressionGenerator.hourly(),
   statistics=data_quality_model_monitor.baseline_statistics(),
   constraints=data_quality_model_monitor.suggested_constraints(),
   schedule_cron_expression=CronExpressionGenerator.hourly(),
   enable_cloudwatch_metrics=True,
   endpoint_input=EndpointInput(
        endpoint_name=endpoint_name,
        destination="/opt/ml/processing/input/endpoint",
   )
)

Surveillance de la qualité des données pour les tâches de transformation par lots

Pour planifier une surveillance de la qualité des données pour une tâche de transformation par lots, transmettez votre instance BatchTransformInput à l'argument batch_transform_input de votre instance DefaultModelMonitor, comme indiqué dans l'exemple de code suivant :

from sagemaker.model_monitor import CronExpressionGenerator
                
data_quality_model_monitor = DefaultModelMonitor(
   role=sagemaker.get_execution_role(),
   ...
)

schedule = data_quality_model_monitor.create_monitoring_schedule(
    monitor_schedule_name=mon_schedule_name,
    batch_transform_input=BatchTransformInput(
        data_captured_destination_s3_uri=s3_capture_upload_path,
        destination="/opt/ml/processing/input",
        dataset_format=MonitoringDatasetFormat.csv(header=False),
    ),
    output_s3_uri=s3_report_path,
    statistics= statistics_path,
    constraints = constraints_path,
    schedule_cron_expression=CronExpressionGenerator.hourly(),
    enable_cloudwatch_metrics=True,
)