Exécuter un pipeline - Amazon SageMaker

Exécuter un pipeline

Une fois que vous avez créé une définition de pipeline à l'aide du kit SDK Python SageMaker, vous pouvez l'envoyer à SageMaker pour lancer votre exécution. Le didacticiel suivant montre comment envoyer un pipeline, lancer une exécution, examiner les résultats de cette exécution et supprimer votre pipeline.

Prerequisites

Pour suivre ce didacticiel, vous devez disposer de la configuration suivante :

  • Une instance de bloc-notes Amazon SageMaker. 

  • Une définition du pipeline SageMaker Pipelines. Ce didacticiel suppose que vous utilisez la définition de pipeline créée en suivant le didacticiel Définition d'un pipeline.

Étape 1 : démarrer le pipeline

Tout d'abord, vous devez démarrer le pipeline.

Pour démarrer le pipeline

  1. Examinez la définition de pipeline JSON pour vous assurer qu'elle est bien formée.

    import json json.loads(pipeline.definition())
  2. Envoyez la définition du pipeline au service SageMaker Pipelines pour créer un pipeline s'il n'existe pas, ou mettez à jour le pipeline s'il existe. Le rôle transmis est utilisé par SageMaker Pipelines pour créer toutes les tâches définies dans les étapes.

    pipeline.upsert(role_arn=role)
  3. Démarrez l'exécution d'un pipeline.

    execution = pipeline.start()

Étape 2 : examiner l'exécution d'un pipeline

Ensuite, vous devez examiner l'exécution du pipeline.

Pour examiner l'exécution d'un pipeline

  1. Décrivez le statut d'exécution du pipeline pour vous assurer qu'il a été créé et démarré avec succès.

    execution.describe()
  2. Attendez que l'exécution soit terminée.

    execution.wait()
  3. Répertoriez les étapes d'exécution et leur état.

    execution.list_steps()

    Le résultat doit être similaire à ce qui suit :

    [{'StepName': 'AbaloneTransform', 'StartTime': datetime.datetime(2020, 11, 21, 2, 41, 27, 870000, tzinfo=tzlocal()), 'EndTime': datetime.datetime(2020, 11, 21, 2, 45, 50, 492000, tzinfo=tzlocal()), 'StepStatus': 'Succeeded', 'CacheHitResult': {'SourcePipelineExecutionArn': ''}, 'Metadata': {'TransformJob': {'Arn': 'arn:aws:sagemaker:us-east-2:111122223333:transform-job/pipelines-cfvy1tjuxdq8-abalonetransform-ptyjoef3jy'}}}, {'StepName': 'AbaloneRegisterModel', 'StartTime': datetime.datetime(2020, 11, 21, 2, 41, 26, 929000, tzinfo=tzlocal()), 'EndTime': datetime.datetime(2020, 11, 21, 2, 41, 28, 15000, tzinfo=tzlocal()), 'StepStatus': 'Succeeded', 'CacheHitResult': {'SourcePipelineExecutionArn': ''}, 'Metadata': {'RegisterModel': {'Arn': 'arn:aws:sagemaker:us-east-2:111122223333:model-package/abalonemodelpackagegroupname/1'}}}, {'StepName': 'AbaloneCreateModel', 'StartTime': datetime.datetime(2020, 11, 21, 2, 41, 26, 895000, tzinfo=tzlocal()), 'EndTime': datetime.datetime(2020, 11, 21, 2, 41, 27, 708000, tzinfo=tzlocal()), 'StepStatus': 'Succeeded', 'CacheHitResult': {'SourcePipelineExecutionArn': ''}, 'Metadata': {'Model': {'Arn': 'arn:aws:sagemaker:us-east-2:111122223333:model/pipelines-cfvy1tjuxdq8-abalonecreatemodel-jl94rai0ra'}}}, {'StepName': 'AbaloneMSECond', 'StartTime': datetime.datetime(2020, 11, 21, 2, 41, 25, 558000, tzinfo=tzlocal()), 'EndTime': datetime.datetime(2020, 11, 21, 2, 41, 26, 329000, tzinfo=tzlocal()), 'StepStatus': 'Succeeded', 'CacheHitResult': {'SourcePipelineExecutionArn': ''}, 'Metadata': {'Condition': {'Outcome': 'True'}}}, {'StepName': 'AbaloneEval', 'StartTime': datetime.datetime(2020, 11, 21, 2, 37, 34, 767000, tzinfo=tzlocal()), 'EndTime': datetime.datetime(2020, 11, 21, 2, 41, 18, 80000, tzinfo=tzlocal()), 'StepStatus': 'Succeeded', 'CacheHitResult': {'SourcePipelineExecutionArn': ''}, 'Metadata': {'ProcessingJob': {'Arn': 'arn:aws:sagemaker:us-east-2:111122223333:processing-job/pipelines-cfvy1tjuxdq8-abaloneeval-zfraozhmny'}}}, {'StepName': 'AbaloneTrain', 'StartTime': datetime.datetime(2020, 11, 21, 2, 34, 55, 867000, tzinfo=tzlocal()), 'EndTime': datetime.datetime(2020, 11, 21, 2, 37, 34, 34000, tzinfo=tzlocal()), 'StepStatus': 'Succeeded', 'CacheHitResult': {'SourcePipelineExecutionArn': ''}, 'Metadata': {'TrainingJob': {'Arn': 'arn:aws:sagemaker:us-east-2:111122223333:training-job/pipelines-cfvy1tjuxdq8-abalonetrain-tavd6f3wdf'}}}, {'StepName': 'AbaloneProcess', 'StartTime': datetime.datetime(2020, 11, 21, 2, 30, 27, 160000, tzinfo=tzlocal()), 'EndTime': datetime.datetime(2020, 11, 21, 2, 34, 48, 390000, tzinfo=tzlocal()), 'StepStatus': 'Succeeded', 'CacheHitResult': {'SourcePipelineExecutionArn': ''}, 'Metadata': {'ProcessingJob': {'Arn': 'arn:aws:sagemaker:us-east-2:111122223333:processing-job/pipelines-cfvy1tjuxdq8-abaloneprocess-mgqyfdujcj'}}}]
  4. Une fois l'exécution de votre pipeline terminée, téléchargez le fichier evaluation.json résultant d'Amazon S3 pour examiner le rapport.

    evaluation_json = sagemaker.s3.S3Downloader.read_file("{}/evaluation.json".format( step_eval.arguments["ProcessingOutputConfig"]["Outputs"][0]["S3Output"]["S3Uri"] )) json.loads(evaluation_json)

Étape 3 : remplacer les paramètres par défaut d'une exécution de pipeline

Vous pouvez exécuter d'autres exécutions du pipeline en spécifiant différents paramètres de pipeline pour remplacer les valeurs par défaut.

Pour remplacer les paramètres par défaut

  1. Créez l'exécution du pipeline. Cela permet de lancer une autre exécution de pipeline sur un type d'instance optimisé pour le calcul et de définir l'état d'approbation du modèle sur « Approved » (Approuvé). Cela signifie que la version du package de modèles générée par l'étape RegisterModel est automatiquement prête à être déployée via des pipelines CI/CD, par exemple avec SageMaker Projects. Pour de plus amples informations, veuillez consulter . Automatiser les MLOps avec des projets SageMaker.

    execution = pipeline.start( parameters=dict( ProcessingInstanceType="ml.c5.xlarge", ModelApprovalStatus="Approved", ) )
  2. Attendez que l'exécution soit terminée.

    execution.wait()
  3. Répertoriez les étapes d'exécution et leur état.

    execution.list_steps()
  4. Une fois l'exécution de votre pipeline terminée, téléchargez le fichier evaluation.json résultant d'Amazon S3 pour examiner le rapport.

    evaluation_json = sagemaker.s3.S3Downloader.read_file("{}/evaluation.json".format( step_eval.arguments["ProcessingOutputConfig"]["Outputs"][0]["S3Output"]["S3Uri"] )) json.loads(evaluation_json)

Étape 4 : arrêter et supprimer une exécution de pipeline

Lorsque vous n'avez plus besoin de votre pipeline, vous pouvez arrêter toutes les exécutions en cours et supprimer le pipeline.

Pour arrêter et supprimer une exécution de pipeline

  1. Arrêtez l'exécution du pipeline.

    execution.stop()
  2. Supprimez le pipeline.

    pipeline.delete()