Ajout d’une étape

Dans l’interface utilisateur de glisser-déposer de Pipelines, vous pouvez utiliser une étape Exécuter le code pour exécuter votre propre code en tant qu’étape de pipeline. Vous pouvez charger une fonction, un script ou un bloc-notes Python à exécuter dans le cadre de votre pipeline. Vous devez utiliser cette étape si vous souhaitez orchestrer une tâche ML personnalisée qui tire parti des fonctionnalités avancées de SageMaker AI ou d’autres services AWS.

L’étape Exécuter le code charge des fichiers dans votre compartiment Amazon S3 par défaut pour Amazon SageMaker AI. Ce compartiment ne dispose pas des autorisations de partage des ressources entre origines multiples (CORS). Pour en savoir plus sur la configuration d’autorisations CORS, consultez Exigence CORS pour les données d’image d’entrée.

L’étape Exécuter le code utilise une tâche d’entraînement Amazon SageMaker pour exécuter votre code. Assurez-vous que votre rôle IAM dispose des autorisations d’API sagemaker:DescribeTrainingJob et sagemaker:CreateTrainingJob. Pour en apprendre davantage sur toutes les autorisations requises pour Amazon SageMaker AI et sur la manière de les configurer, consultez Autorisations d’API Amazon SageMaker AI : référence des actions, des autorisations et des ressources.

Pour ajouter une étape d’exécution de code à un pipeline à l’aide du concepteur de pipeline, procédez comme suit :

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, consultez Données de traitement et Modèles d’évaluation.

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, consultez Entraînement d’un modèle avec Amazon SageMaker AI.

Une étape d’entraînement nécessite un estimateur, ainsi que des entrées de données d’entraînement et de validation.

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, consultez Réglage de modèle automatique avec SageMaker AI.

La tâche de réglage est associée à l’expérience SageMaker AI pour le pipeline, avec les tâches d’entraînement créées sous forme d’essais. Pour plus d’informations, consultez 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, consultez 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 AI, consultez sagemaker-pipelines-tuning-step.ipynb.

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

from sagemaker.workflow.pipeline_context import PipelineSession

from sagemaker.tuner import HyperparameterTuner
from sagemaker.inputs import TrainingInput
from sagemaker.workflow.steps import TuningStep

tuner = HyperparameterTuner(..., sagemaker_session=PipelineSession())
    
step_tuning = TuningStep(
    name = "HPTuning",
    step_args = tuner.fit(inputs=TrainingInput(s3_data="s3://amzn-s3-demo-bucket/my-data"))
)

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, consultez la documentation sagemaker.workflow.steps.TuningStep.

Le peaufinage entraîne un modèle de fondation pré-entraîné d’Amazon SageMaker JumpStart sur un nouveau jeu de données. Ce processus, également connu sous le nom d’apprentissage par transfert, peut produire des modèles précis avec des jeux de données plus petits et moins de temps d’entraînement. Lorsque vous optimisez un modèle, vous pouvez utiliser le jeu de données par défaut ou choisir vos propres données. Pour en savoir plus sur le peaufinage d’un modèle de fondation à partir de JumpStart, consultez Affiner un modèle.

L’étape de peaufinage utilise une tâche d’entraînement Amazon SageMaker pour personnaliser votre modèle. Assurez-vous que votre rôle IAM dispose des autorisations d’API sagemaker:DescribeTrainingJob et sagemaker:CreateTrainingJob nécessaires pour exécuter la tâche de peaufinage dans votre pipeline. Pour en apprendre davantage sur les autorisations requises pour Amazon SageMaker AI et sur la manière de les configurer, consultez Autorisations d’API Amazon SageMaker AI : référence des actions, des autorisations et des ressources.

Pour ajouter une étape Ajuster le modèle à votre pipeline à l’aide de l’éditeur par glisser-déposer, procédez comme suit :

Utilisez l'API AutoML pour créer une tâche AutoML afin d'entraîner automatiquement un modèle. Pour plus d'informations sur les tâches AutoML, consultez Automatiser le développement de modèles avec Amazon SageMaker Autopilot.

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

from sagemaker.workflow.pipeline_context import PipelineSession
from sagemaker.workflow.automl_step import AutoMLStep

pipeline_session = PipelineSession()

auto_ml = AutoML(...,
    role="<role>",
    target_attribute_name="my_target_attribute_name",
    mode="ENSEMBLING",
    sagemaker_session=pipeline_session) 

input_training = AutoMLInput(
    inputs="s3://amzn-s3-demo-bucket/my-training-data",
    target_attribute_name="my_target_attribute_name",
    channel_type="training",
)
input_validation = AutoMLInput(
    inputs="s3://amzn-s3-demo-bucket/my-validation-data",
    target_attribute_name="my_target_attribute_name",
    channel_type="validation",
)

step_args = auto_ml.fit(
    inputs=[input_training, input_validation]
)

step_automl = AutoMLStep(
    name="AutoMLStep",
    step_args=step_args,
)     

Obtenir la meilleure version de modèle

L’étape AutoML entraîne automatiquement plusieurs modèles candidats. Obtenez le modèle avec la meilleure métrique d’objectif de la tâche AutoML à l’aide de la méthode get_best_auto_ml_model comme suit. Vous devez également utiliser un role IAM pour accéder aux artefacts du modèle.

best_model = step_automl.get_best_auto_ml_model(role=<role>)

Pour plus d’informations, consultez l’étape AutoML dans le SDK Python de SageMaker.

Utilisez une étape ModelStep pour créer ou enregistrer un modèle SageMaker AI. Pour plus d’informations sur les conditions de ModelStep, consultez la documentation sagemaker.workflow.steps.CreateModelStep.

Création d’un modèle

Vous utilisez une étape ModelStep pour créer un modèle SageMaker AI. Une étape ModelStep nécessite des artefacts de modèle et des informations à propos du type d’instance SageMaker AI que vous devez utiliser pour créer le modèle. Pour plus d’informations sur les modèles SageMaker AI, consultez Entraînement d’un modèle avec Amazon SageMaker AI.

L’exemple suivant montre comment créer une définition ModelStep.

from sagemaker.workflow.pipeline_context import PipelineSession
from sagemaker.model import Model
from sagemaker.workflow.model_step import ModelStep

step_train = TrainingStep(...)
model = Model(
    image_uri=pytorch_estimator.training_image_uri(),
    model_data=step_train.properties.ModelArtifacts.S3ModelArtifacts,
    sagemaker_session=PipelineSession(),
    role=role,
)

step_model_create = ModelStep(
   name="MyModelCreationStep",
   step_args=model.create(instance_type="ml.m5.xlarge"),
)

Enregistrement d’un modèle

Vous pouvez utiliser une étape ModelStep pour enregistrer un élément sagemaker.model.Model ou sagemaker.pipeline.PipelineModel auprès du registre des 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, consultez Déploiement de l’enregistrement de modèles à l’aide du registre de modèles.

L’exemple suivant montre comment créer une ModelStep qui enregistre un PipelineModel.

import time

from sagemaker.workflow.pipeline_context import PipelineSession
from sagemaker.sklearn import SKLearnModel
from sagemaker.xgboost import XGBoostModel

pipeline_session = PipelineSession()

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='1.0-1',
   role=role,
   sagemaker_session=pipeline_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=pipeline_session,
   role=role
)

from sagemaker.workflow.model_step import ModelStep
from sagemaker import PipelineModel

pipeline_model = PipelineModel(
   models=[sklearn_model, xgboost_model],
   role=role,sagemaker_session=pipeline_session,
)

register_model_step_args = pipeline_model.register(
    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',
)

step_model_registration = ModelStep(
   name="AbaloneRegisterModel",
   step_args=register_model_step_args,
)

Vous utilisez une étape Créer un modèle pour créer un modèle SageMaker AI. Pour plus d’informations sur les modèles SageMaker AI, consultez Entraînement d’un modèle dans Amazon SageMaker.

Une étape de création de modèle nécessite des artefacts de modèle et des informations à propos du type d’instance SageMaker AI que vous devez utiliser pour créer le modèle. Les exemples suivants montrent comment créer une définition d’étape Créer un modèle. Pour plus d’informations sur les conditions de l’étape Créer un modèle, consultez la documentation sagemaker.workflow.steps.CreateModelStep.

L’étape Enregistrer le modèle enregistre un modèle dans le registre de modèles SageMaker.

Dans le concepteur de pipeline, utilisez l’étape Déployer le modèle (point de terminaison) pour déployer votre modèle sur un point de terminaison. Vous pouvez créer un nouveau point de terminaison ou utiliser un point de terminaison existant. L’inférence en temps réel est idéale pour les charges de travail d’inférence où vous avez des exigences en temps réel, interactives et à faible latence. Vous pouvez déployer votre modèle sur les services d’hébergement SageMaker AI et obtenir un point de terminaison en temps réel qui peut être utilisé pour l’inférence. Ces points de terminaison sont entièrement gérés et prennent en charge l’autoscaling. Pour en apprendre davantage sur l’inférence en temps réel dans SageMaker AI, consultez Inférence en temps réel.

Avant d’ajouter une étape de déploiement de modèle à votre pipeline, assurez-vous que votre rôle IAM possède les autorisations suivantes :

Pour en apprendre davantage sur toutes les autorisations requises pour SageMaker AI et sur la façon de les configurer, consultez Autorisations d’API Amazon SageMaker AI : référence des actions, des autorisations et des ressources.

Pour ajouter une étape de déploiement du modèle à votre pipeline dans l’éditeur par glisser-déposer, procédez comme suit :

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, consultez Transformations par lots avec 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 de transformation. Pour plus d’informations sur les conditions requises pour l’étape de transformation, consultez la documentation sagemaker.workflow.steps.TransformStep.

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 :

Utilisez une étape Callback pour ajouter des processus supplémentaires et des services AWS dans votre flux de travail qui ne sont pas directement fournis par Amazon SageMaker Pipelines. Lorsqu’une étape de Callback s’exécute, la procédure suivante se produit :

Pour plus d’informations sur les conditions de l’étape Callback, consultez la documentation sagemaker.workflow.callback_step.CallbackStep. Pour obtenir une solution complète, consultez Étendre les pipelines Amazon SageMaker pour inclure des étapes personnalisées en utilisant des étapes de rappel.

L’exemple suivant montre 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 AI 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 Callback est en cours d’exécution.

Lorsque vous appelez StopPipelineExecution sur un processus de pipeline avec une étape Callback en cours d’exécution, Pipelines envoie un message Amazon SQS à la file d’attente SQS. 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 consommateur de messages Amazon SQS pour effectuer toutes les actions nécessaires (par exemple, le nettoyage des ressources) dès la réception du message. Ajoutez ensuite un appel à SendPipelineExecutionStepSuccess ou à SendPipelineExecutionStepFailure.

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

Vous utilisez une étape Lambda pour exécuter une fonction AWS Lambda. Vous pouvez exécuter une fonction Lambda existante, ou SageMaker AI peut créer et exécuter une nouvelle fonction Lambda. Si vous choisissez d’utiliser une fonction Lambda existante, elle doit être dans la même Région AWS que le pipeline SageMaker AI. Pour voir un bloc-notes qui montre comment utiliser une étape Lambda dans un pipeline SageMaker AI, consultez sagemaker-pipelines-lambda-step.ipynb.

SageMaker AI fournit la classe sagemaker.lambda_helper.Lambda pour créer, mettre à jour, invoquer 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 :

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"
    ]
)

Entrées et sorties

Si votre fonction Lambda a des entrées ou des sorties, elles doivent également être définies dans votre étape Lambda.

Lors de la définition de l’étape Lambda, inputs doit être un dictionnaire de paires clé-valeur. Chaque valeur du dictionnaire inputs doit être de type primitif (chaîne, entier ou flottante). Les objets imbriqués ne sont pas pris en charge. Si elle n'est pas définie, la valeur inputs est définie par défaut sur None.

La valeur outputs doit être une liste de clés. Ces clés font référence à un dictionnaire défini dans la sortie de la fonction Lambda. Comme inputs, ces clés doivent être de type primitif et les objets imbriqués ne sont pas pris en charge.

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 arrêtez le processus pendant que la fonction Lambda est en cours d’exécution, le pipeline attend que la fonction se termine ou que le délai d’expiration soit atteint. Cela dépend de la première éventualité. Le processus s’arrête alors. 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.

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. Vous pouvez ensuite générer et enregistrer vos références avec la méthode model.register() et transmettre la sortie de cette méthode à Étape du modèle en utilisant step_args. 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. Par conséquent, il n’est pas nécessaire de faire 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 de modèles. L’étape ClarifyCheck utilise le conteneur prédéfini SageMaker Clarify. Ce conteneur 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 plus d’informations, consultez Conteneurs SageMaker Clarify prédéfinis.

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.

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

L’étape ClarifyCheck lance une tâche de traitement qui exécute le conteneur prédéfini SageMaker AI 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. Ces fonctions d’assistance sont alignées sur la façon dont la tâche de traitement SageMaker Clarify calcule pour vérifier le biais du modèle, le biais des données ou l’explicabilité du modèle. Pour plus d’informations, consultez 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 :

Pour plus d’informations, consultez Calcul de référence, détection de dérive et cycle de vie avec les étapes ClarifyCheck et QualityCheck dans Amazon SageMaker Pipelines.

Utilisation des références

Le cas échéant, vous pouvez spécifier l’élément model_package_group_name pour localiser la référence existante. Ensuite, l’étape ClarifyCheck tire (pull) l’élément DriftCheckBaselines sur le dernier package de modèle approuvé dans le groupe de packages de modèle.

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, consultez sagemaker.workflow.steps.ClarifyCheckStep dans le kit Amazon SageMaker AI SDK pour Python. Pour voir un bloc-notes Amazon SageMaker Studio Classic qui montre comment utiliser l’étape ClarifyCheck dans Pipelines, consultez sagemaker-pipeline-model-monitor-clarify-steps.ipynb.

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

Utilisez 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. Vous pouvez ensuite générer et enregistrer vos références avec la méthode model.register() et transmettre la sortie de cette méthode à Étape du modèle en utilisant step_args. ]

Model Monitor peut utiliser ces références pour la vérification de dérive 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 de modèles. Cette étape QualityCheck utilise le conteneur prédéfini Amazon SageMaker AI Model Monitor. Ce conteneur propose diverses fonctionnalités de surveillance de modèles, y compris la suggestion de contraintes, la génération de statistiques et la validation de contraintes par rapport à une référence. Pour plus d’informations, consultez Conteneur préconçu Amazon SageMaker Model Monitor.

Configuration de l’étape QualityCheck

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

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

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. Ces fonctions d’assistance 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, consultez Création d’une référence et Création d’une 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 :

Pour plus d’informations, consultez Calcul de référence, détection de dérive et cycle de vie avec les étapes ClarifyCheck et QualityCheck dans Amazon SageMaker 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. Vous pouvez également spécifier model_package_group_name, et l’étape QualityCheck tire (pull) l’élément DriftCheckBaselines sur le dernier package de modèle approuvé dans le groupe de packages de modèle.

Lorsque vous spécifiez ce qui suit, 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.

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

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

Utilisez l’étape EMR d’Amazon SageMaker Pipelines pour :

Pour plus d’informations sur Amazon EMR, consultez Bien démarrer avec Amazon EMR.

L’étape EMR nécessite que l’élément EMRStepConfig inclue l’emplacement du fichier JAR utilisé par le cluster Amazon EMR et tous les arguments à transmettre. Vous fournissez également l’ID de cluster Amazon EMR si vous souhaitez exécuter l’étape sur un cluster EMR en cours d’exécution. Vous pouvez également transmettre la configuration du cluster pour exécuter l’étape EMR sur un cluster qu’il crée, gère et résilie pour vous. Les sections suivantes incluent des exemples et des liens vers des exemples de blocs-notes illustrant les deux méthodes.

Lancement d’une nouvelle tâche sur un cluster Amazon EMR en cours d’exécution

Si vous lancez une nouvelle tâche sur un cluster Amazon EMR en cours d’exécution, transmettez l’ID du cluster sous forme de chaîne à l’argument cluster_id de EMRStep. L’exemple suivant illustre cette procédure.

from sagemaker.workflow.emr_step import EMRStep, EMRStepConfig

emr_config = EMRStepConfig(
    jar="jar-location", # required, path to jar file used
    args=["--verbose", "--force"], # optional list of arguments to pass to the jar
    main_class="com.my.Main1", # optional main class, this can be omitted if jar above has a manifest 
    properties=[ # optional list of Java properties that are set when the step runs
    {
        "key": "mapred.tasktracker.map.tasks.maximum",
        "value": "2"
    },
    {
        "key": "mapreduce.map.sort.spill.percent",
        "value": "0.90"
   },
   {
       "key": "mapreduce.tasktracker.reduce.tasks.maximum",
       "value": "5"
    }
  ]
)

step_emr = EMRStep (
    name="EMRSampleStep", # required
    cluster_id="j-1ABCDEFG2HIJK", # include cluster_id to use a running cluster
    step_config=emr_config, # required
    display_name="My EMR Step",
    description="Pipeline step to execute EMR job"
)

Pour obtenir un exemple de bloc-notes qui vous guide à travers un exemple complet, consultez Pipelines EMR Step With Running EMR Cluster.

Lancement d’une nouvelle tâche sur un nouveau cluster Amazon EMR

Pour lancer une nouvelle tâche sur un nouveau cluster créé pour vous par EMRStep, fournissez la configuration de votre cluster sous forme de dictionnaire. Le dictionnaire doit avoir la même structure qu’une demande RunJobFlow. Toutefois, n’incluez pas les champs suivants dans la configuration de votre cluster :

Tous les autres arguments RunJobFlow peuvent être utilisés dans votre configuration de cluster. Pour plus de détails sur la syntaxe des demandes, consultez RunJobFlow.

L’exemple suivant transmet une configuration de cluster à une définition d’étape EMR. Cela invite l’étape à lancer une nouvelle tâche sur un nouveau cluster EMR. Dans cet exemple, la configuration du cluster EMR inclut des spécifications pour les nœuds primaires et principaux du cluster EMR. Pour plus d’informations sur les types de nœuds Amazon EMR, consultez Comprendre les types de nœuds : nœuds primaires, principaux et de tâches.

from sagemaker.workflow.emr_step import EMRStep, EMRStepConfig

emr_step_config = EMRStepConfig(
    jar="jar-location", # required, path to jar file used
    args=["--verbose", "--force"], # optional list of arguments to pass to the jar
    main_class="com.my.Main1", # optional main class, this can be omitted if jar above has a manifest 
    properties=[ # optional list of Java properties that are set when the step runs
    {
        "key": "mapred.tasktracker.map.tasks.maximum",
        "value": "2"
    },
    {
        "key": "mapreduce.map.sort.spill.percent",
        "value": "0.90"
   },
   {
       "key": "mapreduce.tasktracker.reduce.tasks.maximum",
       "value": "5"
    }
  ]
)

# include your cluster configuration as a dictionary
emr_cluster_config = {
    "Applications": [
        {
            "Name": "Spark", 
        }
    ],
    "Instances":{
        "InstanceGroups":[
            {
                "InstanceRole": "MASTER",
                "InstanceCount": 1,
                "InstanceType": "m5.2xlarge"
            },
            {
                "InstanceRole": "CORE",
                "InstanceCount": 2,
                "InstanceType": "m5.2xlarge"
            }
        ]
    },
    "BootstrapActions":[],
    "ReleaseLabel": "emr-6.6.0",
    "JobFlowRole": "job-flow-role",
    "ServiceRole": "service-role"
}

emr_step = EMRStep(
    name="emr-step",
    cluster_id=None,
    display_name="emr_step",
    description="MyEMRStepDescription",
    step_config=emr_step_config,
    cluster_config=emr_cluster_config
)

Pour obtenir un exemple de bloc-notes qui vous guide à travers un exemple complet, consultez Pipelines EMR Step With Cluster Lifecycle Management.

Utilisez une étape NotebookJobStep pour exécuter votre tâche de bloc-notes SageMaker de manière non interactive en tant qu’étape de pipeline. Si vous générez votre pipeline dans l’interface utilisateur de glisser-déposer de Pipelines, utilisez l’Étape Exécuter le code pour exécuter votre bloc-notes. Pour plus d’informations sur les tâches de bloc-notes SageMaker, consultez Tâches de bloc-notes SageMaker.

Un élément NotebookJobStep nécessite au minimum un bloc-notes d’entrée, une URI d’image et un nom de noyau. Pour plus d’informations sur les exigences relatives aux étapes de tâche de bloc-notes et les autres paramètres que vous pouvez définir pour personnaliser votre étape, consultez sagemaker.workflow.steps.NotebookJobStep.

L’exemple suivant utilise un minimum d’arguments pour définir une NotebookJobStep.

from sagemaker.workflow.notebook_job_step import NotebookJobStep


notebook_job_step = NotebookJobStep(
    input_notebook=input_notebook,
    image_uri=image_uri,
    kernel_name=kernel_name
)

Votre étape de pipeline NotebookJobStep est traitée comme une tâche de bloc-notes SageMaker. Par conséquent, suivez le statut d’exécution dans le tableau de bord des tâches de bloc-notes de l’interface utilisateur de Studio Classic en incluant des balises spécifiques avec l’argument tags. Pour plus d’informations sur les balises à inclure, consultez Visualisation de vos tâches de bloc-notes dans le tableau de bord de l’interface utilisateur de Studio.

De plus, si vous planifiez votre tâche de bloc-notes à l’aide du kit SageMaker Python SDK, vous ne pouvez spécifier que certaines images pour exécuter votre tâche de bloc-notes. Pour plus d’informations, consultez Contraintes d’image pour les tâches de bloc-notes du kit SageMaker AI Python SDK.

Utilisez une étape Échec pour arrêter une exécution d’Amazon SageMaker Pipelines lorsqu’une condition ou un état souhaité n’est pas atteint. L’étape Échec vous permet également d’entrer un message d’erreur personnalisé, indiquant la cause de l’échec de l’exécution du pipeline.

Limitations liées à l’utilisation de l’étape Échec

Vous pouvez créer l’étape Échec sous la forme d’une chaîne de texte statique. Vous pouvez également utiliser les Paramètres du pipeline, une opération de jointure, ou d’autres propriétés d’étape pour créer un message d’erreur plus informatif si vous utilisez le kit SDK.