Définition d’un pipeline

1. Ouvrez la console Studio en suivant les instructions figurant dans Lancez Amazon SageMaker Studio.

2. Dans le volet de navigation de gauche, sélectionnez Pipelines.

3. Choisissez Créer.

4. Choisissez Vide.

5. Dans la barre latérale gauche, choisissez Traiter les données et faites-le glisser vers le canevas.

6. Dans le canevas, choisissez l’étape Traiter les données que vous avez ajoutée.

7. Pour ajouter un jeu de données d’entrée, choisissez Ajouter sous Données (entrée) dans la barre latérale droite et sélectionnez un jeu de données.

8. Pour ajouter un emplacement pour enregistrer les jeux de données de sortie, choisissez Ajouter sous Données (sortie) dans la barre latérale droite et naviguez jusqu’à la destination.

9. Complétez les champs restants dans la barre latérale droite. Pour plus d'informations sur les champs de ces onglets, consultez sagemaker.workflow.steps. ProcessingStep.

1. Dans la barre latérale gauche, choisissez Entraîner le modèle et faites-le glisser vers le canevas.

2. Dans le canevas, choisissez l’étape Entraîner le modèle que vous avez ajoutée.

3. Pour ajouter un jeu de données d’entrée, choisissez Ajouter sous Données (entrée) dans la barre latérale droite et sélectionnez un jeu de données.

4. Pour choisir un emplacement où enregistrer les artefacts de votre modèle, entrez un URI Amazon S3 dans le champ Emplacement (URI S3) ou choisissez Parcourir S3 pour accéder à l’emplacement de destination.

5. Complétez les champs restants dans la barre latérale droite. Pour plus d'informations sur les champs de ces onglets, consultez sagemaker.workflow.steps. TrainingStep.

6. Cliquez sur le curseur et faites le glisser de l’étape Traiter les données que vous avez ajoutée dans la section précédente vers l’étape Entraîner le modèle pour créer une périphérie reliant les deux étapes.

1. Dans la barre latérale gauche, choisissez Enregistrer le modèle et faites-le glisser vers le canevas.

2. Dans le canevas, choisissez l’étape Enregistrer le modèle que vous avez ajoutée.

3. Pour sélectionner un modèle à enregistrer, choisissez Ajouter sous Modèle (entrée).

4. Choisissez Créer un groupe de modèles pour ajouter votre modèle à un nouveau groupe de modèles.

5. Complétez les champs restants dans la barre latérale droite. Pour plus d'informations sur les champs de ces onglets, consultez sagemaker.workflow.step_collections. RegisterModel.

6. Cliquez sur le curseur et faites-le glisser depuis l’étape Entraîner le modèle que vous avez ajoutée dans la section précédente jusqu’à l’étape Enregistrer le modèle pour créer une périphérie reliant les deux étapes.

1. Dans la barre latérale gauche, choisissez Déployer le modèle (point de terminaison) et faites-le glisser vers le canevas.

2. Dans le canevas, choisissez l’étape Déployer le modèle (point de terminaison) que vous avez ajoutée.

3. Pour choisir un modèle à déployer, choisissez Ajouter sous Modèle (entrée).

4. Choisissez la case d’option Créer un point de terminaison pour créer un nouveau point de terminaison.

5. Entrez un nom et une description pour votre point de terminaison.

6. Cliquez sur le curseur et faites-le glisser depuis l’étape Enregistrer le modèle que vous avez ajoutée dans la section précédente jusqu’à l’étape Déployer le modèle (point de terminaison) pour créer une périphérie reliant les deux étapes.

7. Complétez les champs restants dans la barre latérale droite.

Pour télécharger le jeu de données

1. Téléchargez le jeu de données dans le compartiment Amazon S3 par défaut de votre compte.

   !mkdir -p data
   local_path = "data/abalone-dataset.csv"
   
   s3 = boto3.resource("s3")
   s3.Bucket(f"sagemaker-servicecatalog-seedcode-{region}").download_file(
       "dataset/abalone-dataset.csv",
       local_path
   )
   
   base_uri = f"s3://{default_bucket}/abalone"
   input_data_uri = sagemaker.s3.S3Uploader.upload(
       local_path=local_path, 
       desired_s3_uri=base_uri,
   )
   print(input_data_uri)

2. Téléchargez un deuxième jeu de données pour la transformation par lots après la création de votre modèle.

   local_path = "data/abalone-dataset-batch.csv"
   
   s3 = boto3.resource("s3")
   s3.Bucket(f"sagemaker-servicecatalog-seedcode-{region}").download_file(
       "dataset/abalone-dataset-batch",
       local_path
   )
   
   base_uri = f"s3://{default_bucket}/abalone"
   batch_data_uri = sagemaker.s3.S3Uploader.upload(
       local_path=local_path, 
       desired_s3_uri=base_uri,
   )
   print(batch_data_uri)

Pour créer une étape de traitement

1. Créez une instance d'un processeur à transmettre à l'étape de traitement.

   from sagemaker.core.processing import Processor
   from sagemaker.core import image_uris
   
   framework_version = "0.23-1"
   
   sklearn_image_uri = image_uris.retrieve(
       framework="sklearn",
       region=region,
       version=framework_version,
       instance_type="ml.m5.xlarge"
   )
   
   sklearn_processor = Processor(
       image_uri=sklearn_image_uri,
       instance_type="ml.m5.xlarge",
       instance_count=processing_instance_count,
       base_job_name="sklearn-abalone-process",
       sagemaker_session=pipeline_session,
       role=role,
   )

2. Créer une étape de traitement.

   from sagemaker.core.processing import ProcessingInput, ProcessingOutput
   from sagemaker.mlops.workflow.steps import ProcessingStep
   
   processor_args = sklearn_processor.run(
       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",
   ) 
   
   step_process = ProcessingStep(name="AbaloneProcess", step_args=processor_args)

Pour définir une étape d’entraînement

1. Spécifiez le chemin d'accès au modèle dans lequel vous souhaitez enregistrer les modèles de l'entraînement.

   model_path = f"s3://{default_bucket}/AbaloneTrain"

2. Configurez un estimateur d'entraînement pour l'algorithme XGBoost et le jeu de données en entrée.

   from sagemaker.train import ModelTrainer
   from sagemaker.train.configs import Compute, OutputDataConfig
   from sagemaker.core import image_uris
   
   image_uri = image_uris.retrieve(
       framework="xgboost",
       region=region,
       version="1.0-1",
       py_version="py3",
       instance_type="ml.m5.xlarge"
   )
   xgb_train = ModelTrainer(
       training_image=image_uri,
       compute=Compute(instance_type="ml.m5.xlarge", instance_count=1),
       output_data_config=OutputDataConfig(s3_output_path=model_path),
       hyperparameters={
           "objective": "reg:linear",
           "num_round": "50",
           "max_depth": "5",
           "eta": "0.2",
           "gamma": "4",
           "min_child_weight": "6",
           "subsample": "0.7",
           "silent": "0"
       },
       sagemaker_session=pipeline_session,
       role=role,
   )

3. Créez un TrainingStep à l'aide de l'instance d'entraînement et des propriétés duProcessingStep.

   from sagemaker.train.configs import InputData
   from sagemaker.mlops.workflow.steps import TrainingStep
   
   train_args = xgb_train.train(
       input_data_config=[
           InputData(
               channel_name="train",
               data_source=step_process.properties.ProcessingOutputConfig.Outputs[
                   "train"
               ].S3Output.S3Uri,
               content_type="text/csv"
           ),
           InputData(
               channel_name="validation",
               data_source=step_process.properties.ProcessingOutputConfig.Outputs[
                   "validation"
               ].S3Output.S3Uri,
               content_type="text/csv"
           )
       ],
   )
   
   step_train = TrainingStep(name="AbaloneTrain", step_args=train_args)

Pour définir une étape de traitement pour l’évaluation du modèle

1. Créez une instance de Processor pour évaluation.

   from sagemaker.core.processing import Processor
   
   script_eval = Processor(
       image_uri=image_uri,
       command=["python3"],
       instance_type="ml.m5.xlarge",
       instance_count=1,
       base_job_name="script-abalone-eval",
       sagemaker_session=pipeline_session,
       role=role,
   )

2. Créez un ProcessingStep à l'aide de l'instance du processeur, des canaux d'entrée et de sortie et du evaluation.py script.

   from sagemaker.core.workflow.properties import PropertyFile
   
   evaluation_report = PropertyFile(
       name="EvaluationReport",
       output_name="evaluation",
       path="evaluation.json"
   )
   
   eval_args = script_eval.run(
       inputs=[
           ProcessingInput(
               source=step_train.properties.ModelArtifacts.S3ModelArtifacts,
               destination="/opt/ml/processing/model"
           ),
           ProcessingInput(
               source=step_process.properties.ProcessingOutputConfig.Outputs[
                   "test"
               ].S3Output.S3Uri,
               destination="/opt/ml/processing/test"
           )
       ],
       outputs=[
           ProcessingOutput(output_name="evaluation", source="/opt/ml/processing/evaluation"),
       ],
       code="abalone/evaluation.py",
   )
   
   step_eval = ProcessingStep(
       name="AbaloneEval",
       step_args=eval_args,
       property_files=[evaluation_report],
   )

Pour définir un CreateModelStep pour la transformation par lots

• Créez un modèle d' SageMaker IA et une étape de modélisation.

  from sagemaker.serve import ModelBuilder
  from sagemaker.mlops.workflow.model_step import ModelStep
  
  model_builder = ModelBuilder(
      image_uri=image_uri,
      s3_model_data_url=step_train.properties.ModelArtifacts.S3ModelArtifacts,
      role_arn=role,
      sagemaker_session=pipeline_session,
  )
  
  step_create_model = ModelStep(
      name="AbaloneCreateModel",
      step_args=model_builder.build(),
  )

Pour définir un TransformStep pour effectuer une transformation par lots

• Créez une instance de transformateur et unTransformStep.

  from sagemaker.core.transformer import Transformer
  from sagemaker.mlops.workflow.steps import TransformStep
  
  transformer = Transformer(
      model_name=step_create_model.properties.ModelName,
      instance_type="ml.m5.xlarge",
      instance_count=1,
      output_path=f"s3://{default_bucket}/AbaloneTransform",
      sagemaker_session=pipeline_session,
  )
  
  transform_args = transformer.transform(data=batch_data)
  
  step_transform = TransformStep(
      name="AbaloneTransform",
      step_args=transform_args,
  )

Pour définir une étape d'enregistrement de modèle afin de créer un package de modèles

• Créez une étape d'enregistrement modèle.

  from sagemaker.core.model_metrics import MetricsSource, ModelMetrics 
  from sagemaker.mlops.workflow.model_step import ModelStep
  from sagemaker.serve import ModelBuilder
  
  model_metrics = ModelMetrics(
      model_statistics=MetricsSource(
          s3_uri="{}/evaluation.json".format(
              step_eval.arguments["ProcessingOutputConfig"]["Outputs"][0]["S3Output"]["S3Uri"]
          ),
          content_type="application/json"
      )
  )
  
  model_builder = ModelBuilder(
      image_uri=image_uri,
      s3_model_data_url=step_train.properties.ModelArtifacts.S3ModelArtifacts,
      role_arn=role,
      sagemaker_session=pipeline_session,
  )
  
  register_args = model_builder.register(
      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
  )
  
  step_register = ModelStep(
      name="AbaloneRegisterModel",
      step_args=register_args,
  )

Pour définir une étape de condition pour vérifier la précision du modèle

• Définissez une condition et construisez-en uneConditionStep.

  from sagemaker.core.workflow.conditions import ConditionLessThanOrEqualTo
  from sagemaker.mlops.workflow.condition_step import ConditionStep
  from sagemaker.core.workflow.functions import JsonGet
  
  cond_lte = ConditionLessThanOrEqualTo(
      left=JsonGet(
          step_name=step_eval.name,
          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=[], 
  )

Pour créer un pipeline

1. Définissez les éléments suivants pour votre pipeline :name, parameters, et steps.

   from sagemaker.mlops.workflow.pipeline import Pipeline
   
   pipeline_name = f"AbalonePipeline"
   pipeline = Pipeline(
       name=pipeline_name,
       parameters=[
           processing_instance_count,
           model_approval_status,
           input_data,
           batch_data,
       ],
       steps=[step_process, step_train, step_eval, step_cond],
   )

2. (Facultatif) Examinez la définition de pipeline JSON pour vous assurer qu’elle est bien formée.

   import json
   
   json.loads(pipeline.definition())