Étapes du pipeline - Amazon SageMaker

Étapes du pipeline

SageMaker Pipelines se compose d'étapes. Ces étapes définissent les actions effectuées par le pipeline et les relations entre les étapes utilisant les propriétés.

Types d'étapes

La section suivante décrit les exigences de chaque type d'étape et fournit un exemple d'implémentation de l'étape. Ce ne sont pas des implémentations fonctionnelles, car elles ne fournissent pas les ressources et les entrées nécessaires. Pour obtenir un didacticiel qui met en œuvre ces étapes, veuillez consulter Créer et gérer SageMaker Pipelines.

Amazon SageMaker Model Building Pipelines prend en charge les types d'étape suivants :

Étape de traitement

Vous utilisez une étape de traitement pour créer une tâche de traitement pour le traitement des données. Pour plus d'informations sur les tâches de traitement, veuillez consulter Données de traitement et Modèles d'évaluation.

Une étape de traitement nécessite un processeur, un script Python qui définit le code de traitement, les sorties pour le traitement et les arguments de tâche. L'exemple suivant montre comment créer une définition ProcessingStep. Pour plus d'informations sur les conditions de l'étape de traitement, veuillez consulter la documentation sagemaker.workflow.steps.ProcessingStep.

from sagemaker.sklearn.processing import SKLearnProcessor sklearn_processor = SKLearnProcessor(framework_version='0.20.0', role=<role>, instance_type='ml.m5.xlarge', instance_count=1)
from sagemaker.processing import ProcessingInput, ProcessingOutput from sagemaker.workflow.steps import ProcessingStep step_process = ProcessingStep( name="AbaloneProcess", processor=sklearn_processor, inputs=[ ProcessingInput(source=<input_data>, destination="/opt/ml/processing/input"), ], outputs=[ ProcessingOutput(output_name="train", source="/opt/ml/processing/train"), ProcessingOutput(output_name="validation", source="/opt/ml/processing/validation"), ProcessingOutput(output_name="test", source="/opt/ml/processing/test") ], code="abalone/preprocessing.py" )

Transmission des paramètres d'exécution

Vous pouvez transmettre des paramètres d'exécution à une étape de traitement à l'aide de la méthode get_run_args du kit SDK Amazon SageMaker Python. Cela vous permet d'utiliser des processeurs en plus de SKLearnProcessor, tels que PySparkProcessor et SparkJarProcessor.

L'exemple suivant montre comment transmettre des paramètres d'exécution d'un processeur PySpark à une ProcessingStep.

from sagemaker.spark.processing import PySparkProcessor pyspark_processor = PySparkProcessor(framework_version='2.4', role=<role>, instance_type='ml.m5.xlarge', instance_count=1)
from sagemaker.processing import ProcessingInput, ProcessingOutput run_args = pyspark_processor.get_run_args( "preprocess.py", inputs=[ ProcessingInput(source=<input_data>, destination="/opt/ml/processing/input"), ], outputs=[ ProcessingOutput(output_name="train", source="/opt/ml/processing/train"), ProcessingOutput(output_name="validation", source="/opt/ml/processing/validation"), ProcessingOutput(output_name="test", source="/opt/ml/processing/test") ], arguments=None )
from sagemaker.workflow.steps import ProcessingStep step_process = ProcessingStep( name="AbaloneProcess", processor=pyspark_processor, inputs=run_args.inputs, outputs=run_args.outputs, job_arguments=run_args.arguments, code=run_args.code )

Étape d'entraînement

Vous utilisez une étape d'entraînement pour créer une tâche d'entraînement afin d'entraîner un modèle. Pour plus d'informations sur les tâches d'entraînement, veuillez consulter Entraînement d'un modèle avec Amazon SageMaker.

Une étape d'entraînement nécessite un estimateur, ainsi que des entrées de données d'entraînement et de validation. L'exemple suivant montre comment créer une définition TrainingStep. Pour de plus amples informations sur les conditions requises pour l'étape d'entraînement, veuillez consulter la documentation sagemaker.workflow.steps.TrainingStep.

from sagemaker.inputs import TrainingInput from sagemaker.workflow.steps import TrainingStep step_train = TrainingStep( name="TrainAbaloneModel", estimator=xgb_train, inputs={ "train": TrainingInput( s3_data=step_process.properties.ProcessingOutputConfig.Outputs[ "train" ].S3Output.S3Uri, content_type="text/csv" ), "validation": TrainingInput( s3_data=step_process.properties.ProcessingOutputConfig.Outputs[ "validation" ].S3Output.S3Uri, content_type="text/csv" ) } )

Étape de réglage

Vous utilisez une étape de réglage pour créer une tâche de réglage d'hyperparamètres, également appelé optimisation des hyperparamètres (HPO). Une tâche de réglage d'hyperparamètres exécute plusieurs tâches d'entraînement, chacune produisant une version de modèle. Pour plus d'informations sur le réglage d'hyperparamètres, veuillez consulter Réglage de modèle automatique avec SageMaker.

La tâche de réglage est associée à l'expérience SageMaker pour le pipeline, avec les tâches d'entraînement créées sous forme d'essais. Pour de plus amples informations, veuillez consulter . Intégration d'Experiments.

Une étape de réglage nécessite un HyperParameterTuner et les entrées d'entraînement. Vous pouvez entraîner à nouveau les tâches de réglage précédentes en spécifiant le paramètre warm_start_config du HyperparameterTuner. Pour plus d'informations sur le réglage d'hyperparamètres et le démarrage à chaud, veuillez consulter Exécution d'une tâche de réglage des hyperparamètres avec démarrage à chaud.

Vous utilisez la méthode get_top_model_s3_uri de la classe sagemaker.workflow.steps.TuningStep pour obtenir l'artefact du modèle à partir de l'une des versions de modèle les plus performantes. Pour obtenir un bloc-notes qui montre comment utiliser une étape de réglage dans un pipeline SageMaker, veuillez consulter sagemaker-pipelines-tuning-step.ipynb.

Important

Les étapes de réglage ont été introduites dans le kit SDK Amazon SageMaker Python version 2.48.0 et Amazon SageMaker Studio version 3.8.0. Vous devez mettre à jour Studio avant d'utiliser une étape de réglage ou le DAG du pipeline ne s'affiche pas. Pour mettre à jour Studio, veuillez consulter Mettre à jour SageMaker Studio.

L'exemple suivant montre comment créer une définition TuningStep.

from sagemaker.tuner import HyperparameterTuner from sagemaker.inputs import TrainingInput from sagemaker.workflow.steps import TuningStep step_tuning = TuningStep( name = "HPTuning", tuner = HyperparameterTuner(...), inputs = TrainingInput(s3_data=input_path) )

Obtenir la meilleure version de modèle

L'exemple suivant montre comment obtenir la meilleure version de modèle à partir de la tâche de réglage à l'aide de la méthode get_top_model_s3_uri. Au plus, les 50 versions les plus performantes sont disponibles et classées selon HyperParameterTuningJobJective. L'argument top_k est un index dans les versions, où top_k=0 est la version la plus performante et top_k=49 est la version la moins performante.

best_model = Model( image_uri=image_uri, model_data=step_tuning.get_top_model_s3_uri( top_k=0, s3_bucket=sagemaker_session.default_bucket() ), ... )

Pour plus d'informations sur les conditions de l'étape de réglage, veuillez consulter la documentation sagemaker.workflow.steps.TuningStep.

Étape CreateModel

Vous utilisez une étape CreateModel pour créer un modèle SageMaker. Pour de plus amples informations sur les modèles SageMaker, veuillez consulter Entraînement d'un modèle dans Amazon SageMaker.

Une étape CreateModel nécessite des artefacts de modèle et des informations sur le type d'instance SageMaker que vous devez utiliser pour créer le modèle. L'exemple suivant montre comment créer une définition d'étape CreateModel. Pour plus d'informations sur les conditions de l'étape CreateModel, veuillez consulter la documentation sagemaker.workflow.steps.CreateModelStep.

from sagemaker.workflow.steps import CreateModelStep step_create_model = CreateModelStep( name="AbaloneCreateModel", model=best_model, inputs=inputs )

Étape RegisterModel

Vous utilisez une étape RegisterModel pour enregistrer un sagemaker.model.Model ou un sagemaker.pipeline.PipelineModel avec le registre de modèles Amazon SageMaker. Un PipelineModel représente un pipeline d'inférence, qui est un modèle composé d'une séquence linéaire de conteneurs qui traitent les demandes d'inférence.

Pour savoir comment enregistrer un modèle, veuillez consulter Enregistrer et déployer des modèles avec Model Registry. Pour plus d'informations sur les conditions de l'étape RegisterModel, veuillez consulter la documentation sagemaker.workflow.step_collections.RegisterModel.

L'exemple suivant montre comment créer une étape RegisterModel qui enregistre un PipelineModel.

import time from sagemaker.sklearn import SKLearnModel from sagemaker.xgboost import XGBoostModel code_location = 's3://{0}/{1}/code'.format(bucket_name, prefix) sklearn_model = SKLearnModel(model_data=processing_step.properties.ProcessingOutputConfig.Outputs['model'].S3Output.S3Uri, entry_point='inference.py', source_dir='sklearn_source_dir/', code_location=code_location, framework_version='0.20.0', role=role, sagemaker_session=sagemaker_session, py_version='py3') xgboost_model = XGBoostModel(model_data=training_step.properties.ModelArtifacts.S3ModelArtifacts, entry_point='inference.py', source_dir='xgboost_source_dir/', code_location=code_location, framework_version='0.90-2', py_version='py3', sagemaker_session=sagemaker_session, role=role) from sagemaker.workflow.step_collections import RegisterModel from sagemaker import PipelineModel pipeline_model = PipelineModel(models=[sklearn_model,xgboost_model],role=role,sagemaker_session=sagemaker_session) step_register = RegisterModel( name="AbaloneRegisterModel", model=pipeline_model, content_types=["application/json"], response_types=["application/json"], inference_instances=["ml.t2.medium", "ml.m5.xlarge"], transform_instances=["ml.m5.xlarge"], model_package_group_name='sipgroup', )

Si le model n'est pas fourni, l'étape RegisterModel nécessite un estimateur, comme illustré dans l'exemple suivant.

from sagemaker.workflow.step_collections import RegisterModel step_register = RegisterModel( name="AbaloneRegisterModel", estimator=xgb_train, model_data=step_train.properties.ModelArtifacts.S3ModelArtifacts, content_types=["text/csv"], response_types=["text/csv"], inference_instances=["ml.t2.medium", "ml.m5.xlarge"], transform_instances=["ml.m5.xlarge"], model_package_group_name=model_package_group_name, approval_status=model_approval_status, model_metrics=model_metrics )

Étape de transformation

Pour exécuter des inférences sur un jeu de données entier, vous utilisez une étape de transformation pour une transformation par lots. Pour plus d'informations sur la transformation par lots, veuillez consulter Exécution de transformations par lots avec des pipelines d'inférence.

Une étape de transformation nécessite un transformateur et les données sur lesquelles exécuter la transformation par lots. L'exemple suivant montre comment créer une définition d'étape Transform. Pour de plus amples informations sur les conditions requises pour l'étape Transform, veuillez consulter la documentation sagemaker.workflow.steps.TransformStep.

from sagemaker.inputs import TransformInput from sagemaker.workflow.steps import TransformStep step_transform = TransformStep( name="AbaloneTransform", transformer=transformer, inputs=TransformInput(data=batch_data) )

Étape de condition

Vous utilisez une étape de condition pour évaluer la condition des propriétés de l'étape afin d'évaluer quelle action doit être effectuée ensuite dans le pipeline.

Une étape de condition nécessite une liste de conditions, une liste d'étapes à exécuter si la condition a pour valeur true et une liste des étapes à exécuter si la condition a pour valeur false. L'exemple suivant montre comment créer une définition d'étape Condition. Pour plus d'informations sur les conditions de l'étape Condition, veuillez consulter la documentation sagemaker.workflow.condition_step.ConditionStep.

Limites

  • SageMaker Pipelines ne prend pas en charge l'utilisation des étapes de condition imbriquées. Vous ne pouvez pas transmettre une étape de condition comme entrée pour une autre étape de condition.

  • Une étape de condition ne peut pas utiliser des étapes identiques dans les deux branches. Si vous avez besoin de la même fonctionnalité d'étape dans les deux branches, dupliquez l'étape et donnez-lui un nom différent.

from sagemaker.workflow.conditions import ConditionLessThanOrEqualTo from sagemaker.workflow.condition_step import ( ConditionStep, JsonGet ) cond_lte = ConditionLessThanOrEqualTo( left=JsonGet( step=step_eval, property_file=evaluation_report, json_path="regression_metrics.mse.value" ), right=6.0 ) step_cond = ConditionStep( name="AbaloneMSECond", conditions=[cond_lte], if_steps=[step_register, step_create_model, step_transform], else_steps=[] )

Étape de rappel

Vous utilisez une étape de rappel pour incorporer des processus supplémentaires et des services AWS dans votre flux qui ne sont pas directement fournis par Amazon SageMaker Model Building Pipelines. Lorsqu'une étape de rappel s'exécute, la procédure suivante se produit :

  • SageMaker Pipelines envoie un message à une file d'attente Amazon Simple Queue Service (Amazon SQS) indiquée par le client. Le message contient un jeton généré par SageMaker PipeLines et une liste de paramètres d'entrée fournie par le client. Après avoir envoyé le message, SageMaker Pipelines attend une réponse du client.

  • Le client récupère le message dans la file d'attente Amazon SQS et démarre son processus personnalisé.

  • Une fois le processus terminé, le client appelle l'une des API suivantes et soumet le jeton généré par SageMaker Pipelines :

  • L'appel d'API fait en sorte que SageMaker Pipelines poursuive le processus du pipeline ou fait échouer le processus.

Pour plus d'informations sur les conditions de l'étape Callback, veuillez consulter la documentation sagemaker.workflow.callback_step.CallbackStep. Pour obtenir une solution complète, veuillez consulter Extend SageMaker Pipelines to include custom steps using callback steps.

Important

Les étapes de rappel ont été introduites dans le kit SDK Amazon SageMaker Python version 2.45.0 et Amazon SageMaker Studio version 3.6.2. Vous devez mettre à jour Studio avant d'utiliser une étape de rappel ou le DAG du pipeline ne s'affiche pas. Pour mettre à jour Studio, veuillez consulter Mettre à jour SageMaker Studio.

L'exemple suivant illustre une mise en œuvre de la procédure précédente.

from sagemaker.workflow.callback_step import CallbackStep step_callback = CallbackStep( name="MyCallbackStep", sqs_queue_url="https://sqs.us-east-2.amazonaws.com/012345678901/MyCallbackQueue", inputs={...}, outputs=[...] ) callback_handler_code = ' import boto3 import json def handler(event, context): sagemaker_client=boto3.client("sagemaker") for record in event["Records"]: payload=json.loads(record["body"]) token=payload["token"] # Custom processing # Call SageMaker to complete the step sagemaker_client.send_pipeline_execution_step_success( CallbackToken=token, OutputParameters={...} ) '

Comportement d'arrêt

Un processus de pipeline ne s'arrête pas lorsqu'une étape de rappel est en cours d'exécution.

Lorsque vous appelez StopPipelineExecution sur le processus d'un pipeline avec une étape de rappel en cours d'exécution, SageMaker Pipelines envoie un message Amazon SQS supplémentaire à la file d'attente SQS spécifiée. Le corps du message SQS contient un champ Status (Statut) qui est défini sur Stopping. L'exemple suivant montre le corps d'un message SQS.

{ "token": "26vcYbeWsZ", "pipelineExecutionArn": "arn:aws:sagemaker:us-east-2:012345678901:pipeline/callback-pipeline/execution/7pinimwddh3a", "arguments": { "number": 5, "stringArg": "some-arg", "inputData": "s3://sagemaker-us-west-2-012345678901/abalone/abalone-dataset.csv" }, "status": "Stopping" }

Vous devez ajouter une logique à votre client de message Amazon SQS pour effectuer toutes les actions nécessaires (par exemple, le nettoyage des ressources) dès la réception du message suivi d'un appel à SendPipelineExecutionStepSuccess ou SendPipelineExecutionStepFailure.

Ce n'est que lorsque SageMaker Pipelines reçoit l'un de ces appels qu'il arrête le processus du pipeline.

Étape Lambda

Vous utilisez une étape Lambda pour exécuter une fonction AWS Lambda. Vous pouvez exécuter une fonction Lambda existante, ou SageMaker peut créer et exécuter une nouvelle fonction Lambda. Pour voir un bloc-notes qui montre comment utiliser une étape Lambda dans un pipeline SageMaker, veuillez consulter sagemaker-pipelines-lambda-step.ipynb.

Important

Les étapes Lambda ont été introduites dans le kit SDK Amazon SageMaker Python version 2.51.0 et Amazon SageMaker Studio version 3.9.1. Vous devez mettre à jour Studio avant d'utiliser une étape Lambda ou le DAG du pipeline ne s'affiche pas. Pour mettre à jour Studio, veuillez consulter Mettre à jour SageMaker Studio.

SageMaker fournit la classe sagemaker.lambda_helper.Lambda pour créer, mettre à jour, appeler et supprimer des fonctions Lambda. Lambda comporte la signature suivante.

Lambda( function_arn, # Only required argument to invoke an existing Lambda function # The following arguments are required to create a Lambda function: function_name, execution_role_arn, zipped_code_dir, # Specify either zipped_code_dir and s3_bucket, OR script s3_bucket, # S3 bucket where zipped_code_dir is uploaded script, # Path of Lambda function script handler, # Lambda handler specified as "lambda_script.lambda_handler" timeout, # Maximum time the Lambda function can run before the lambda step fails ... )

La classe sagemaker.workflow.lambda_step.LambdaStep a un argument lambda_func de type Lambda. Pour appeler une fonction Lambda existante, la seule exigence est de fournir l'Amazon Resource Name (ARN) de la fonction à function_arn. Si vous ne définissez aucune valeur pour function_arn, vous devez spécifier handler et l'un des éléments suivants :

  • zipped_code_dir – Chemin de la fonction Lambda zippée

    s3_bucket – Compartiment Amazon S3 où zipped_code_dir doit être téléchargé

  • script – Chemin d'accès du fichier script de la fonction Lambda

L'exemple suivant montre comment créer une définition d'étape Lambda qui appelle une fonction Lambda existante.

from sagemaker.workflow.lambda_step import LambdaStep from sagemaker.lambda_helper import Lambda step_lambda = LambdaStep( name="ProcessingLambda", lambda_func=Lambda( function_arn="arn:aws:lambda:us-west-2:012345678910:function:split-dataset-lambda" ), inputs={ s3_bucket = s3_bucket, data_file = data_file }, outputs=[ "train_file", "test_file" ] )

L'exemple suivant montre comment créer une définition d'étape Lambda qui appelle une fonction Lambda existante.

from sagemaker.workflow.lambda_step import LambdaStep from sagemaker.lambda_helper import Lambda step_lambda = LambdaStep( name="ProcessingLambda", lambda_func=Lambda( function_name="split-dataset-lambda", execution_role_arn=execution_role_arn, script="lambda_script.py", handler="lambda_script.lambda_handler", ... ), inputs={ s3_bucket = s3_bucket, data_file = data_file }, outputs=[ "train_file", "test_file" ] )

Délai d'expiration et comportement d'arrêt

La classe Lambda a un argument timeout qui spécifie la durée maximale d'exécution de la fonction Lambda. La valeur par défaut est de 120 secondes, avec une valeur maximum de 10 minutes. Si la fonction Lambda est en cours d'exécution lorsque le délai d'expiration est atteint, l'étape Lambda échoue. Cependant, la fonction Lambda continue de s'exécuter.

Un processus de pipeline ne peut pas être arrêté pendant qu'une étape Lambda est en cours d'exécution, car la fonction Lambda invoquée par l'étape Lambda ne peut pas être arrêtée. Si vous tentez d'arrêter le processus pendant que la fonction Lambda est en cours d'exécution, le pipeline attend que la fonction Lambda se termine ou que le délai d'expiration soit atteint, selon la première éventualité, puis s'arrête. Si la fonction Lambda se termine, le statut de processus du pipeline est Stopped. Si le délai d'expiration est atteint, le statut de processus du pipeline est Failed.

Étape ClarifyCheck

Vous pouvez utiliser l'étape ClarifyCheck afin d'effectuer des vérifications de dérive de référence par rapport aux références précédentes pour l'analyse de biais et l'explicabilité de modèle. Ainsi, vous pouvez générer et enregistrer vos références afin de vérifier la dérive dans l'étape RegisterModel avec le modèle entraîné à travers les pipelines dans la phase de création du modèle. Ces références pour la vérification de la dérive peuvent être utilisées par Amazon SageMaker Model Monitor pour les points de terminaison de votre modèle, de sorte que vous n'avez pas besoin d'effectuer une suggestion de référence séparément. L'étape ClarifyCheck peut également extraire des références pour la vérification de dérive à partir du registre des modèles. L'étape ClarifyCheck exploite le conteneur prédéfini Amazon SageMaker Clarify qui fournit diverses fonctions de surveillance de modèles, y compris des fonctions de suggestion de contraintes et de validation de contraintes par rapport à une référence donnée. Pour de plus amples informations, veuillez consulter Mise en route avec un conteneur SageMaker Clarify.

Configuration de l'étape ClarifyCheck

Vous pouvez configurer l'étape ClarifyCheck pour effectuer l'un des types de vérification suivants chaque fois qu'il est utilisé dans un pipeline.

  • Vérification de biais des données

  • Vérification de biais de modèle

  • Vérification d'explicabilité de modèle

Pour ce faire, définissez le paramètre clarify_check_config avec l'une des valeurs de type de vérification suivantes :

  • DataBiasCheckConfig

  • ModelBiasCheckConfig

  • ModelExplainabilityCheckConfig

L'étape ClarifyCheck lance une tâche de traitement qui exécute le conteneur prédéfini SageMaker Clarify et nécessite des configurations dédiées pour la vérification et la tâche de traitement. ClarifyCheckConfig et CheckJobConfig sont des fonctions d'assistance pour ces configurations qui sont alignées sur la façon dont la tâche de traitement SageMaker Clarify calcule pour vérifier le biais de modèle, le biais de données ou l'explicabilité de modèle. Pour de plus amples informations, veuillez consulter . Exécuter les tâches de traitement SageMaker Clarify pour l'analyse des biais et l'explicabilité.

Contrôle des comportements d'étape pour la vérification de dérive

L'étape ClarifyCheck nécessite les deux indicateurs booléens suivants pour le contrôle de son comportement :

  • skip_check : ce paramètre indique si la vérification de dérive par rapport à la référence précédente est ignorée ou non. S'il est défini sur False, la référence précédente du type de contrôle configuré doit être disponible.

  • register_new_baseline : ce paramètre indique si une référence recalculée est accessible via la propriété BaselineUsedForDriftCheckConstraints de l'étape. S'il est défini sur False, la référence précédente du type de contrôle configuré doit également être disponible. Vous pouvez y accéder via la propriété BaselineUsedForDriftCheckConstraints.

Pour de plus amples informations, veuillez consulter . Calcul de référence, détection de dérive et cycle de vie avec les étapes ClarifyCheck et QualityCheck dans Amazon SageMaker Model Building Pipelines.

Utilisation des références

Vous pouvez éventuellement spécifier le model_package_group_name pour localiser la référence existante et l'étape ClarifyCheck extrait les DriftCheckBaselines sur le dernier package de modèles approuvé dans le groupe de modèles. Vous pouvez également fournir une référence précédente via le paramètre supplied_baseline_constraints. Si vous spécifiez le model_package_group_name et les supplied_baseline_constraints, l'étape ClarifyCheck utilise la référence spécifiée par le paramètre supplied_baseline_constraints.

Pour plus d'informations sur l'utilisation des exigences de l'étape ClarifyCheck, veuillez consulter sagemaker.workflow.steps.ClarifyCheckStep pour le kit SDK Python SageMaker Amazon SageMaker. Pour voir un bloc-notes Amazon SageMaker Studio qui montre comment utiliser l'étape ClarifyCheck dans SageMaker Pipelines, veuillez consulter sagemaker-pipeline-model-monitor-clarify-steps.ipynb.

Exemple Créer une étape ClarifyCheck pour la vérification du biais de données

from sagemaker.workflow.check_job_config import CheckJobConfig from sagemaker.workflow.clarify_check_step import DataBiasCheckConfig, ClarifyCheckStep from sagemaker.workflow.execution_variables import ExecutionVariables check_job_config = CheckJobConfig( role=role, instance_count=1, instance_type="ml.c5.xlarge", volume_size_in_gb=120, sagemaker_session=sagemaker_session, ) data_bias_data_config = DataConfig( s3_data_input_path=step_process.properties.ProcessingOutputConfig.Outputs["train"].S3Output.S3Uri, s3_output_path=Join(on='/', values=['s3:/', your_bucket, base_job_prefix, ExecutionVariables.PIPELINE_EXECUTION_ID, 'databiascheckstep']), label=0, dataset_type="text/csv", s3_analysis_config_output_path=data_bias_analysis_cfg_output_path, ) data_bias_config = BiasConfig( label_values_or_threshold=[15.0], facet_name=[8], facet_values_or_threshold=[[0.5]] ) data_bias_check_config = DataBiasCheckConfig( data_config=data_bias_data_config, data_bias_config=data_bias_config, )h data_bias_check_step = ClarifyCheckStep( name="DataBiasCheckStep", clarify_check_config=data_bias_check_config, check_job_config=check_job_config, skip_check=False, register_new_baseline=False supplied_baseline_constraints="s3://sagemaker-us-west-2-111122223333/baseline/analysis.json", model_package_group_name="MyModelPackageGroup" )

Étape QualityCheck

Vous pouvez utiliser l'étape QualityCheck pour effectuer des suggestions de référence et des vérifications de dérive par rapport à une référence précédente pour la qualité des données ou la qualité du modèle dans un pipeline. Ainsi, vous pouvez générer et enregistrer vos références afin de vérifier la dérive dans l'étape Étape RegisterModel avec le modèle entraîné à travers les pipelines dans la phase de création du modèle. Ces références pour la vérification de dérive peuvent être utilisées par Model Monitor pour les points de terminaison de votre modèle, de sorte que vous n'avez pas besoin d'effectuer une suggestion de référence séparément. L'étape QualityCheck peut également extraire des références pour la vérification de dérive à partir du registre des modèles. L'étape QualityCheck exploite le conteneur prédéfini Amazon SageMaker Model Monitor qui fournit diverses fonctions de surveillance de modèles, y compris des fonctions de suggestion de contraintes, de génération de statistiques et de validation de contraintes par rapport à une référence. Pour de plus amples informations, veuillez consulter . Conteneur préconçu Amazon SageMaker Model Monitor.

Configuration de l'étape QualityCheck

Vous pouvez configurer l'étape QualityCheck pour effectuer l'un des types de vérification suivants chaque fois qu'il est utilisé dans un pipeline.

  • Vérification de la qualité des données

  • Vérification de la qualité du modèle

Pour ce faire, définissez le paramètre quality_check_config avec l'une des valeurs de type de vérification suivantes :

  • DataQualityCheckConfig

  • ModelQualityCheckConfig

L'étape QualityCheck lance une tâche de traitement qui exécute le conteneur prédéfini Model Monitor et nécessite des configurations dédiées pour la vérification et la tâche de traitement. QualityCheckConfig et CheckJobConfig sont des fonctions d'assistance pour ces configurations qui sont alignées sur la façon dont Model Monitor crée une référence pour la surveillance de la qualité du modèle ou de la qualité des données. Pour plus d'informations sur les suggestions de référence Model Monitor, veuillez consulter Création d'une référence et Créer une tâche de référence de qualité des modèles.

Contrôle des comportements d'étape pour la vérification de dérive

L'étape QualityCheck nécessite les deux indicateurs booléens suivants pour le contrôle de son comportement :

  • skip_check : ce paramètre indique si la vérification de dérive par rapport à la référence précédente est ignorée ou non. S'il est défini sur False, la référence précédente du type de contrôle configuré doit être disponible.

  • register_new_baseline : ce paramètre indique si une référence recalculée est accessible via les propriétés BaselineUsedForDriftCheckConstraints et BaselineUsedForDriftCheckStatistics de l'étape. S'il est défini sur False, la référence précédente du type de contrôle configuré doit également être disponible. Vous pouvez y accéder via les propriétés BaselineUsedForDriftCheckConstraints et BaselineUsedForDriftCheckStatistics.

Pour de plus amples informations, veuillez consulter . Calcul de référence, détection de dérive et cycle de vie avec les étapes ClarifyCheck et QualityCheck dans Amazon SageMaker Model Building Pipelines.

Utilisation des références

Vous pouvez spécifier une référence précédente directement via les paramètres supplied_baseline_statistics et supplied_baseline_constraints, ou vous pouvez simplement spécifier le model_package_group_name pour que l'étape QualityCheck extraie les DriftCheckBaselines sur le dernier package de modèles approuvé dans le groupe de modèles. Lorsque vous spécifiez le model_package_group_name, les supplied_baseline_constraints et les supplied_baseline_statistics, l'étape QualityCheck utilise la référence spécifiée par les supplied_baseline_constraints et les supplied_baseline_statistics sur le type de vérification de l'étape QualityCheck que êtes en train d'exécuter.

Pour plus d'informations sur l'utilisation des exigences de l'étape QualityCheck, veuillez consulter sagemaker.workflow.steps.QualityCheckStep pour le kit SDK Python SageMaker Amazon SageMaker. Pour voir un bloc-notes Amazon SageMaker Studio qui montre comment utiliser l'étape QualityCheck dans SageMaker Pipelines, veuillez consulter sagemaker-pipeline-model-monitor-quality-steps.ipynb.

Exemple Créer une étape QualityCheck pour la vérification de la qualité des données

from sagemaker.workflow.check_job_config import CheckJobConfig from sagemaker.workflow.quality_check_step import DataQualityCheckConfig, QualityCheckStep from sagemaker.workflow.execution_variables import ExecutionVariables check_job_config = CheckJobConfig( role=role, instance_count=1, instance_type="ml.c5.xlarge", volume_size_in_gb=120, sagemaker_session=sagemaker_session, ) data_quality_check_config = DataQualityCheckConfig( baseline_dataset=step_process.properties.ProcessingOutputConfig.Outputs["train"].S3Output.S3Uri, dataset_format=DatasetFormat.csv(header=False, output_columns_position="START"), output_s3_uri=Join(on='/', values=['s3:/', your_bucket, base_job_prefix, ExecutionVariables.PIPELINE_EXECUTION_ID, 'dataqualitycheckstep']) ) data_quality_check_step = QualityCheckStep( name="DataQualityCheckStep", skip_check=False, register_new_baseline=False, quality_check_config=data_quality_check_config, check_job_config=check_job_config, supplied_baseline_statistics="s3://sagemaker-us-west-2-555555555555/baseline/statistics.json", supplied_baseline_constraints="s3://sagemaker-us-west-2-555555555555/baseline/constraints.json", model_package_group_name="MyModelPackageGroup" )

Propriétés de l'étape

L'attribut properties est utilisé pour ajouter des dépendances de données entre les étapes du pipeline. Ces dépendances de données sont ensuite utilisées par SageMaker Pipelines pour construire le DAG à partir de la définition du pipeline. Ces propriétés peuvent être référencées en tant que valeurs d'espace réservé et sont résolues lors de l'exécution.

L'attribut properties d'une étape SageMaker Pipelines correspond à l'objet retourné par un appel Describe pour le type de tâche SageMaker correspondant. Pour chaque type de tâche, l'appel Describe renvoie l'objet de réponse suivant :

Dépendance des données entre étapes

Vous définissez la structure de votre DAG en spécifiant les relations des données entre les étapes. Pour créer des dépendances de données entre les étapes, transmettez les propriétés d'une étape comme entrée à une autre étape du pipeline. L'étape recevant l'entrée n'est démarrée qu'après l'étape fournissant l'entrée a terminé l'exécution.

Une dépendance de données utilise la notation JsonPath au format suivant. Ce format traverse le fichier de propriétés JSON, ce qui signifie que vous pouvez ajouter autant d'instances <propiété> que nécessaire pour atteindre la propriété imbriquée souhaitée dans le fichier. Pour plus d'informations sur la notation JsonPath, veuillez consulter le référentiel JsonPath.

<step_name>.properties.<property>.<property>

Ce qui suit montre comment spécifier un compartiment S3 à l'aide de la propriété ProcessingOutputConfig d'une étape de traitement.

step_process.properties.ProcessingOutputConfig.Outputs["train_data"].S3Output.S3Uri

Pour créer la dépendance des données, transmettez le compartiment à une étape d'entraînement comme suit.

step_train = TrainingStep( name="CensusTrain", estimator=sklearn_train, inputs=TrainingInput( s3_data=step_process.properties.ProcessingOutputConfig.Outputs[ "train_data"].S3Output.S3Uri ) )

Dépendance personnalisée entre étapes

Lorsque vous spécifiez une dépendance de données, SageMaker Pipelines fournit la connexion de données entre les étapes. Par ailleurs, une étape peut accéder aux données d'une étape précédente sans utiliser directement SageMaker Pipelines. Dans ce cas, vous pouvez créer une dépendance personnalisée qui indique à SageMaker Pipelines de ne pas démarrer une étape tant qu'une autre étape n'a pas été exécutée. Vous créez une dépendance personnalisée en spécifiant l'attribut DependsOn d'une étape.

À titre d'exemple, ce qui suit définit une étape C qui démarre seulement après que les deux étapes A et B terminent leur exécution.

{ 'Steps': [ {'Name':'A', ...}, {'Name':'B', ...}, {'Name':'C', 'DependsOn': ['A', 'B']} ] }

SageMaker Pipelines lève une exception de validation si la dépendance crée une dépendance cyclique.

L'exemple suivant crée une étape d'entraînement qui démarre après l'exécution d'une étape de traitement.

processing_step = ProcessingStep(...) training_step = TrainingStep(...) training_step.add_depends_on([processing_step])

L'exemple suivant crée une étape d'entraînement qui ne démarre pas tant que l'exécution de deux étapes de traitement différentes n'est pas terminée.

processing_step_1 = ProcessingStep(...) processing_step_2 = ProcessingStep(...) training_step = TrainingStep(...) training_step.add_depends_on([processing_step_1, processing_step_2])

Ce qui suit fournit un autre moyen de créer la dépendance personnalisée.

training_step.add_depends_on([processing_step_1]) training_step.add_depends_on([processing_step_2])

L'exemple suivant crée une étape d'entraînement qui reçoit les entrées d'une étape de traitement et attend que l'exécution d'une autre étape de traitement se termine.

processing_step_1 = ProcessingStep(...) processing_step_2 = ProcessingStep(...) training_step = TrainingStep( ..., inputs=TrainingInput( s3_data=processing_step_1.properties.ProcessingOutputConfig.Outputs[ "train_data" ].S3Output.S3Uri ) training_step.add_depends_on([processing_step_2])

L'exemple suivant montre comment extraire une liste de chaînes des dépendances personnalisées d'une étape.

custom_dependencies = training_step.depends_on

Utiliser une image personnalisée dans une étape

Vous pouvez utiliser l'une des images de Deep Learning Containers SageMaker lorsque vous créez une étape dans votre pipeline.

Vous pouvez également créer une étape à l'aide des applications SageMaker Amazon S3. Une application SageMaker Amazon S3 est une solution groupée tar.gz avec un ou plusieurs scripts Python pouvant s'exécuter dans cette solution groupée. Pour plus d'informations sur la création d'une offre groupée de packages d'applications, veuillez consulter Deploying directly from model artifacts.

Vous pouvez également utiliser votre propre conteneur avec des étapes de pipeline. Comme vous ne pouvez pas créer d'image à partir d'Amazon SageMaker Studio, vous devez créer votre image à l'aide d'une autre méthode avant de l'utiliser avec Amazon SageMaker Model Building Pipelines.

Pour utiliser votre propre conteneur lors de la création des étapes pour votre pipeline, incluez l'URI de l'image dans la définition de l'estimateur. Pour plus d'informations sur l'utilisation de votre propre conteneur avec SageMaker, veuillez consulterUtilisation de conteneurs Docker avec SageMaker.