Définition d’un pipeline

1. Ouvrez la console Studio en suivant les instructions figurant dans Lancement d’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 en savoir plus 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 en savoir plus 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 en savoir plus 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 un répertoire pour le script de traitement.

   !mkdir -p abalone

2. Dans le répertoire /abalone, créez un fichier nommé preprocessing.py avec le contenu suivant. Ce script de prétraitement est transmis à l’étape de traitement pour être exécuté sur les données d’entrée. L’étape d’entraînement utilise ensuite les caractéristiques et les étiquettes d’entraînement prétraitées pour entraîner un modèle. L’étape d’évaluation utilise le modèle entraîné ainsi que les caractéristiques et les étiquettes de test prétraitées pour évaluer le modèle. Le script utilise scikit-learn pour effectuer les opérations suivantes :

   • Compléter les données de catégorie sex manquantes et les encoder pour qu'elles soient adaptées à l'entraînement.

   • Mettre à l'échelle et normaliser tous les champs numériques à l'exception de rings et sex.

   • Diviser les données en jeux de données d'entraînement, de validation et de test.

   %%writefile abalone/preprocessing.py
   import argparse
   import os
   import requests
   import tempfile
   import numpy as np
   import pandas as pd
   
   
   from sklearn.compose import ColumnTransformer
   from sklearn.impute import SimpleImputer
   from sklearn.pipeline import Pipeline
   from sklearn.preprocessing import StandardScaler, OneHotEncoder
   
   
   # Because this is a headerless CSV file, specify the column names here.
   feature_columns_names = [
       "sex",
       "length",
       "diameter",
       "height",
       "whole_weight",
       "shucked_weight",
       "viscera_weight",
       "shell_weight",
   ]
   label_column = "rings"
   
   feature_columns_dtype = {
       "sex": str,
       "length": np.float64,
       "diameter": np.float64,
       "height": np.float64,
       "whole_weight": np.float64,
       "shucked_weight": np.float64,
       "viscera_weight": np.float64,
       "shell_weight": np.float64
   }
   label_column_dtype = {"rings": np.float64}
   
   
   def merge_two_dicts(x, y):
       z = x.copy()
       z.update(y)
       return z
   
   
   if __name__ == "__main__":
       base_dir = "/opt/ml/processing"
   
       df = pd.read_csv(
           f"{base_dir}/input/abalone-dataset.csv",
           header=None, 
           names=feature_columns_names + [label_column],
           dtype=merge_two_dicts(feature_columns_dtype, label_column_dtype)
       )
       numeric_features = list(feature_columns_names)
       numeric_features.remove("sex")
       numeric_transformer = Pipeline(
           steps=[
               ("imputer", SimpleImputer(strategy="median")),
               ("scaler", StandardScaler())
           ]
       )
   
       categorical_features = ["sex"]
       categorical_transformer = Pipeline(
           steps=[
               ("imputer", SimpleImputer(strategy="constant", fill_value="missing")),
               ("onehot", OneHotEncoder(handle_unknown="ignore"))
           ]
       )
   
       preprocess = ColumnTransformer(
           transformers=[
               ("num", numeric_transformer, numeric_features),
               ("cat", categorical_transformer, categorical_features)
           ]
       )
       
       y = df.pop("rings")
       X_pre = preprocess.fit_transform(df)
       y_pre = y.to_numpy().reshape(len(y), 1)
       
       X = np.concatenate((y_pre, X_pre), axis=1)
       
       np.random.shuffle(X)
       train, validation, test = np.split(X, [int(.7*len(X)), int(.85*len(X))])
   
       
       pd.DataFrame(train).to_csv(f"{base_dir}/train/train.csv", header=False, index=False)
       pd.DataFrame(validation).to_csv(f"{base_dir}/validation/validation.csv", header=False, index=False)
       pd.DataFrame(test).to_csv(f"{base_dir}/test/test.csv", header=False, index=False)

3. Créer une instance d'un SKLearnProcessor pour la transmettre à l'étape de traitement.

   from sagemaker.sklearn.processing import SKLearnProcessor
   
   
   framework_version = "0.23-1"
   
   sklearn_processor = SKLearnProcessor(
       framework_version=framework_version,
       instance_type="ml.m5.xlarge",
       instance_count=processing_instance_count,
       base_job_name="sklearn-abalone-process",
       sagemaker_session=pipeline_session,
       role=role,
   )

4. Créer une étape de traitement. Cette étape adopte le SKLearnProcessor, les canaux d’entrée et de sortie, ainsi que le script preprocessing.py que vous avez créé. Ceci est très similaire à la méthode run de l’instance de processeur dans le kit SageMaker AI Python SDK. Le paramètre input_data transmis dans ProcessingStep correspond aux données d’entrée de l’étape elle-même. Ces données d'entrée sont utilisées par l'instance du processeur lors de son exécution.

   Notez les canaux nommés "train, "validation et "test" spécifiés dans la configuration de sortie pour la tâche de traitement. Les Properties de l’étape telles que celles-ci peuvent être utilisées dans les étapes suivantes et correspondre à leurs valeurs d’exécution au moment de l’exécution.

   from sagemaker.processing import ProcessingInput, ProcessingOutput
   from sagemaker.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 pour l’algorithme XGBoost et le jeu de données d’entrée. Le type d’instance d’entraînement est transmis à l’estimateur. Un script d’entraînement standard :

   • charge les données depuis les canaux d’entrée ;

   • configure l’entraînement avec des hyperparamètres ;

   • entraîne un modèle ;

   • enregistre un modèle dans model_dir pour qu’il puisse être hébergé ultérieurement.

   SageMaker AI charge le modèle sur Amazon S3 sous la forme d’un model.tar.gz à la fin de la tâche d’entraînement.

   from sagemaker.estimator import Estimator
   
   
   image_uri = sagemaker.image_uris.retrieve(
       framework="xgboost",
       region=region,
       version="1.0-1",
       py_version="py3",
       instance_type="ml.m5.xlarge"
   )
   xgb_train = Estimator(
       image_uri=image_uri,
       instance_type="ml.m5.xlarge",
       instance_count=1,
       output_path=model_path,
       sagemaker_session=pipeline_session,
       role=role,
   )
   xgb_train.set_hyperparameters(
       objective="reg:linear",
       num_round=50,
       max_depth=5,
       eta=0.2,
       gamma=4,
       min_child_weight=6,
       subsample=0.7,
       silent=0
   )

3. Créez une étape TrainingStep à l’aide de l’instance d’estimateur et des propriétés de l’étape ProcessingStep. Transmettez l’élément S3Uri du canal de sortie "train" et "validation" à l’étape TrainingStep. 

   from sagemaker.inputs import TrainingInput
   from sagemaker.workflow.steps import TrainingStep
   
   
   train_args = xgb_train.fit(
       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"
           )
       },
   )
   
   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 un fichier dans le répertoire /abalone nommé evaluation.py. Ce script est utilisé dans une étape de traitement pour effectuer l'évaluation du modèle. Il prend un modèle entraîné et le jeu de données de test comme entrée, puis produit un fichier JSON contenant des métriques d’évaluation de classification.

   %%writefile abalone/evaluation.py
   import json
   import pathlib
   import pickle
   import tarfile
   import joblib
   import numpy as np
   import pandas as pd
   import xgboost
   
   
   from sklearn.metrics import mean_squared_error
   
   
   if __name__ == "__main__":
       model_path = f"/opt/ml/processing/model/model.tar.gz"
       with tarfile.open(model_path) as tar:
           tar.extractall(path=".")
       
       model = pickle.load(open("xgboost-model", "rb"))
   
       test_path = "/opt/ml/processing/test/test.csv"
       df = pd.read_csv(test_path, header=None)
       
       y_test = df.iloc[:, 0].to_numpy()
       df.drop(df.columns[0], axis=1, inplace=True)
       
       X_test = xgboost.DMatrix(df.values)
       
       predictions = model.predict(X_test)
   
       mse = mean_squared_error(y_test, predictions)
       std = np.std(y_test - predictions)
       report_dict = {
           "regression_metrics": {
               "mse": {
                   "value": mse,
                   "standard_deviation": std
               },
           },
       }
   
       output_dir = "/opt/ml/processing/evaluation"
       pathlib.Path(output_dir).mkdir(parents=True, exist_ok=True)
       
       evaluation_path = f"{output_dir}/evaluation.json"
       with open(evaluation_path, "w") as f:
           f.write(json.dumps(report_dict))

2. Créez une instance de ScriptProcessor qui est utilisée pour créer une ProcessingStep.

   from sagemaker.processing import ScriptProcessor
   
   
   script_eval = ScriptProcessor(
       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,
   )

3. Créez une étape ProcessingStep en utilisant l’instance de processeur, les canaux d’entrée et de sortie et le script evaluation.py. Transmettez-lui :

   • la propriété S3ModelArtifacts issue de l’étape d’entraînement step_train,

   • l’élément S3Uri du canal de sortie "test" de l’étape de traitement step_process.

   Ceci est très similaire à la méthode run de l’instance de processeur dans le kit SageMaker AI Python SDK. 

   from sagemaker.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 une CreateModelStep pour la transformation par lots

1. Créez un modèle SageMaker AI. Transmettez la propriété S3ModelArtifacts depuis l’étape d’entraînement step_train.

   from sagemaker.model import Model
   
   
   model = Model(
       image_uri=image_uri,
       model_data=step_train.properties.ModelArtifacts.S3ModelArtifacts,
       sagemaker_session=pipeline_session,
       role=role,
   )

2. Définissez l’entrée du modèle pour votre modèle SageMaker AI.

   from sagemaker.inputs import CreateModelInput
   
   
   inputs = CreateModelInput(
       instance_type="ml.m5.large",
       accelerator_type="ml.eia1.medium",
   )

3. Créez votre étape CreateModelStep en utilisant l’élément CreateModelInput et l’instance de modèle SageMaker AI que vous avez définie.

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

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

1. Créez une instance de transformateur avec le type d'instance de calcul approprié, le nombre d'instances et l'URI de compartiment Amazon S3 de sortie souhaitée. Transmettez la propriété ModelName depuis l'étape step_create_model CreateModel.

   from sagemaker.transformer import Transformer
   
   
   transformer = Transformer(
       model_name=step_create_model.properties.ModelName,
       instance_type="ml.m5.xlarge",
       instance_count=1,
       output_path=f"s3://{default_bucket}/AbaloneTransform"
   )

2. Créez une TransformStep à l'aide de l'instance de transformateur que vous avez définie et du paramètre de pipeline batch_data.

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

Pour définir une étape RegisterModel pour créer un package de modèles

• Créez une RegisterModel à l'aide de l'instance d'estimateur que vous avez utilisée pour l'étape d'entraînement. Transmettez la propriété S3ModelArtifacts depuis l’étape d’entraînement step_train et spécifiez un ModelPackageGroup. Pipelines crée cet élément ModelPackageGroup pour vous.

  from sagemaker.model_metrics import MetricsSource, ModelMetrics 
  from sagemaker.workflow.step_collections import RegisterModel
  
  
  model_metrics = ModelMetrics(
      model_statistics=MetricsSource(
          s3_uri="{}/evaluation.json".format(
              step_eval.arguments["ProcessingOutputConfig"]["Outputs"][0]["S3Output"]["S3Uri"]
          ),
          content_type="application/json"
      )
  )
  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
  )

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

1. Définissez une condition ConditionLessThanOrEqualToen utilisant la valeur de précision trouvée dans la sortie de l'étape de traitement de l'évaluation du modèle, step_eval. Obtenez cette sortie en utilisant le fichier de propriétés que vous avez indexé à l'étape de traitement et le JSONPath respectif de la valeur d'erreur quadratique moyenne, "mse".

   from sagemaker.workflow.conditions import ConditionLessThanOrEqualTo
   from sagemaker.workflow.condition_step import ConditionStep
   from sagemaker.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
   )

2. Créez une ConditionStep. Transmettez la condition ConditionEquals, puis définissez les étapes d'enregistrement de package de modèle et de transformation par lots comme les étapes suivantes si la condition est satisfaite.

   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. Les noms doivent être uniques au sein d'une paire (account, region).

   Note

   Une étape ne peut apparaître qu'une seule fois dans la liste des étapes du pipeline ou dans les listes d'étapes if/else de l'étape de condition. Elle ne peut pas apparaître dans les deux.

   from sagemaker.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())