Planification des tâches de surveillance

Amazon SageMaker Model Monitor vous permet de surveiller les données collectées à partir de vos points de terminaison en temps réel. Vous pouvez surveiller vos données selon une planification récurrente ou les surveiller une fois, immédiatement. Vous pouvez créer un calendrier de surveillance à l'aide du CreateMonitoringScheduleAPI.

Avec un calendrier de surveillance, SageMaker vous pouvez commencer à traiter des tâches pour analyser les données collectées au cours d'une période donnée. Au cours de la tâche de traitement, SageMaker compare le jeu de données pour l'analyse en cours avec les statistiques de référence et les contraintes que vous fournissez. SageMaker Générez ensuite un rapport de violations. De plus, CloudWatch des métriques sont émises pour chaque caractéristique analysée.

SageMaker fournit un conteneur prédéfini pour effectuer des analyses sur des ensembles de données tabulaires. Vous pouvez également choisir d'apporter votre propre conteneur comme indiqué dans la rubrique Support pour vos propres conteneurs avec Amazon SageMaker Model Monitor.

Vous pouvez créer un calendrier de surveillance des modèles pour votre point de terminaison en temps réel ou votre tâche de transformation par lots. Comparez le trafic en temps réel ou les entrées de tâches par lots par rapport aux ressources de référence (contraintes et statistiques).

Exemple affectations de référence

Dans l'exemple suivant, le jeu de données d'entraînement utilisé pour entraîner le modèle a été chargé sur Amazon S3. S'il est déjà dans Amazon S3, vous pouvez pointer directement dessus.

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

Exemple planification d'une analyse récurrente

Si vous planifiez une surveillance des modèles pour un point de terminaison en temps réel, utilisez des contraintes et des statistiques de référence afin de comparer le trafic en temps réel. L'extrait de code suivant montre le format général que vous utilisez pour planifier une surveillance des modèles pour un point de terminaison en temps réel. Cet exemple planifie le moniteur de modèles pour qu'il s'exécute toutes les heures.

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

mon_schedule_name = 'my-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,
)

Exemple planification d'une analyse ponctuelle

Vous pouvez également planifier l'analyse pour l'exécuter une fois de façon non récurrente en transmettant des arguments tels que les suivants à la méthode create_monitoring_schedule :

    schedule_cron_expression=CronExpressionGenerator.now(),
    data_analysis_start_time="-PT1H",
    data_analysis_end_time="-PT0H",

Dans ces arguments, le paramètre schedule_cron_expression planifie l'analyse pour qu'elle soit exécutée une fois, immédiatement, avec la valeur CronExpressionGenerator.now(). Pour toute planification avec ce paramètre, les paramètres data_analysis_start_time et data_analysis_end_time sont nécessaires. Ces paramètres définissent l'heure de début et de fin d'une fenêtre d'analyse. Définissez ces heures comme des décalages relatifs à l'heure actuelle et utilisez le format de durée ISO 8601. Dans cet exemple, les instants -PT1H et -PT0H définissent une fenêtre entre une heure dans le passé et l'heure actuelle. Avec cette planification, l'analyse évalue uniquement les données collectées au cours de la fenêtre spécifiée.

Exemple planification d'une tâche de transformation par lots

L'extrait de code suivant montre le format général que vous utilisez pour planifier une surveillance des modèles pour une tâche de transformation par lots.

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

mon_schedule_name = 'my-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,
)

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