Les traductions sont fournies par des outils de traduction automatique. En cas de conflit entre le contenu d'une traduction et celui de la version originale en anglais, la version anglaise prévaudra. # Définition d’un pipeline Pour orchestrer vos flux de travail avec Amazon SageMaker Pipelines, vous devez générer un graphe acyclique dirigé (DAG) sous la forme d'une définition de pipeline JSON. Le graphique DAG spécifie les différentes étapes impliquées dans votre processus ML, telles que le prétraitement des données, l’entraînement des modèles, l’évaluation des modèles et le déploiement des modèles, ainsi que les dépendances et le flux de données entre ces étapes. La rubrique suivante explique comment générer une définition de pipeline. Vous pouvez générer votre définition de pipeline JSON à l'aide du SDK SageMaker Python ou de la fonctionnalité visuelle drag-and-drop Pipeline Designer d'Amazon SageMaker Studio. L’image suivante est une représentation du graphique DAG du pipeline que vous créez dans ce didacticiel : ![\[Capture d'écran de l' drag-and-dropinterface visuelle de Pipelines in Studio.\]](http://docs.aws.amazon.com/fr_fr/sagemaker/latest/dg/images/pipelines/pipelines-studio-overview.png) Le pipeline que vous définissez dans les sections suivantes résout un problème de régression visant à déterminer l’âge d’un ormeau en fonction de ses dimensions physiques. Pour un bloc-notes Jupyter exécutable incluant le contenu de ce didacticiel, consultez [Orchestrating Jobs with Amazon SageMaker ](https://sagemaker-examples.readthedocs.io/en/latest/sagemaker-pipelines/tabular/abalone_build_train_deploy/sagemaker-pipelines-preprocess-train-evaluate-batch-transform.html) Model Building Pipelines. **Note** Vous pouvez référencer l'emplacement du modèle en tant que propriété de l'étape d'apprentissage, comme le montre l' end-to-endexemple de [CustomerChurn pipeline](https://github.com/aws-samples/customer-churn-sagemaker-pipelines-sample/blob/main/pipelines/customerchurn/pipeline.py) sur Github. **Topics** ## Définition d’un pipeline (concepteur de pipeline) La procédure pas à pas suivante vous guide à travers les étapes de création d'un pipeline simplifié à l'aide du concepteur de drag-and-drop pipelines. Si vous devez suspendre ou mettre fin à votre session d’édition de pipeline dans le concepteur visuel à un moment quelconque, cliquez sur l’option **Exporter**. Cela vous permet de télécharger la définition actuelle de votre pipeline dans votre environnement local. Plus tard, lorsque vous souhaitez reprendre le processus d’édition du pipeline, vous pouvez importer le même fichier de définition JSON dans le concepteur visuel. ### Création d’une étape de traitement Pour créer une étape de traitement des données, procédez comme suit : 1. Ouvrez la console Studio en suivant les instructions figurant dans [Lancez Amazon SageMaker Studio](studio-updated-launch.md). 1. Dans le volet de navigation de gauche, sélectionnez **Pipelines**. 1. Choisissez **Créer**. 1. Choisissez **Vide**. 1. Dans la barre latérale gauche, choisissez **Traiter les données** et faites-le glisser vers le canevas. 1. Dans le canevas, choisissez l’étape **Traiter les données** que vous avez ajoutée. 1. 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. 1. 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. 1. 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](https://sagemaker.readthedocs.io/en/stable/workflows/pipelines/sagemaker.workflow.pipelines.html#sagemaker.workflow.steps.ProcessingStep). ### Création d’une étape d’entraînement Pour configurer une étape d’entraînement des modèles, procédez comme suit : 1. Dans la barre latérale gauche, choisissez **Entraîner le modèle** et faites-le glisser vers le canevas. 1. Dans le canevas, choisissez l’étape **Entraîner le modèle** que vous avez ajoutée. 1. 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. 1. 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. 1. 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](https://sagemaker.readthedocs.io/en/stable/workflows/pipelines/sagemaker.workflow.pipelines.html#sagemaker.workflow.steps.TrainingStep). 1. 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. ### Création d’un modèle de package avec une étape d’enregistrement du modèle Pour créer un package de modèle avec une étape d’enregistrement de modèle, procédez comme suit : 1. Dans la barre latérale gauche, choisissez **Enregistrer le modèle** et faites-le glisser vers le canevas. 1. Dans le canevas, choisissez l’étape **Enregistrer le modèle** que vous avez ajoutée. 1. Pour sélectionner un modèle à enregistrer, choisissez **Ajouter** sous **Modèle (entrée)**. 1. Choisissez **Créer un groupe de modèles** pour ajouter votre modèle à un nouveau groupe de modèles. 1. 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\$1collections. RegisterModel](https://sagemaker.readthedocs.io/en/stable/workflows/pipelines/sagemaker.workflow.pipelines.html#sagemaker.workflow.step_collections.RegisterModel). 1. 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. ### Déploiement du modèle sur un point de terminaison avec une étape Déployer le modèle (point de terminaison) Pour déployer votre modèle à l’aide d’une étape de déploiement de modèle, procédez comme suit : 1. Dans la barre latérale gauche, choisissez **Déployer le modèle (point de terminaison)** et faites-le glisser vers le canevas. 1. Dans le canevas, choisissez l’étape **Déployer le modèle (point de terminaison)** que vous avez ajoutée. 1. Pour choisir un modèle à déployer, choisissez **Ajouter** sous **Modèle (entrée)**. 1. Choisissez la case d’option **Créer un point de terminaison** pour créer un nouveau point de terminaison. 1. Entrez un **nom** et une **description** pour votre point de terminaison. 1. 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. 1. Complétez les champs restants dans la barre latérale droite. ### Définition des paramètres du pipeline Vous pouvez configurer un ensemble de paramètres de pipeline dont les valeurs peuvent être mises à jour pour chaque exécution. Pour définir les paramètres du pipeline et définir les valeurs par défaut, cliquez sur l’icône représentant un engrenage en bas du concepteur visuel. ### Enregistrement du pipeline Après avoir saisi toutes les informations requises pour créer votre pipeline, cliquez sur **Enregistrer** en bas du concepteur visuel. Cela valide votre pipeline pour détecter toute erreur potentielle lors de l’exécution et vous en informe. L’opération **Enregistrer** ne réussira pas tant que vous n’aurez pas corrigé toutes les erreurs signalées par les contrôles de validation automatisés. Si vous souhaitez reprendre les modifications ultérieurement, vous pouvez enregistrer votre pipeline en cours sous forme de définition JSON dans votre environnement local. Vous pouvez exporter votre pipeline sous la forme d’un fichier de définition JSON en cliquant sur le bouton **Exporter** en bas du concepteur visuel. Plus tard, pour reprendre la mise à jour de votre pipeline, chargez ce fichier de définition JSON en cliquant sur le bouton **Importer**. ## Définition d'un pipeline (SDK SageMaker Python) ### Conditions préalables Pour exécuter le didacticiel suivant, effectuez les opérations suivantes : + Configurez votre instance de bloc-notes comme indiqué dans [Création d’une instance de bloc-notes](https://docs.aws.amazon.com/sagemaker/latest/dg/howitworks-create-ws.html). Cela donne à votre rôle les autorisations nécessaires pour lire et écrire sur Amazon S3, et pour créer des tâches de formation, de transformation par lots et de traitement dans l' SageMaker IA. + Accordez à votre bloc-notes des autorisations pour obtenir et transmettre son propre rôle comme indiqué dans [Modification d’une politique d’autorisations de rôle](https://docs.aws.amazon.com/IAM/latest/UserGuide/roles-managingrole-editing-console.html#roles-modify_permissions-policy). Ajoutez l'extrait JSON suivant pour attacher cette politique à votre rôle. Remplacez `` par l’ARN utilisé pour créer votre instance de bloc-notes. ------ #### [ JSON ] **** ``` { "Version":"2012-10-17", "Statement": [ { "Effect": "Allow", "Action": [ "iam:GetRole", "iam:PassRole" ], "Resource": "arn:aws:iam::111122223333:role/role-name" } ] } ``` ------ + Faites confiance au principal du service d' SageMaker intelligence artificielle en suivant les étapes décrites dans la section [Modification d'une politique de confiance des rôles](https://docs.aws.amazon.com/IAM/latest/UserGuide/roles-managingrole-editing-cli.html#roles-managingrole_edit-trust-policy-cli). Ajoutez le fragment d’instruction suivante à la relation de confiance de votre rôle : ``` { "Sid": "", "Effect": "Allow", "Principal": { "Service": "sagemaker.amazonaws.com" }, "Action": "sts:AssumeRole" } ``` #### Configuration de votre environnement Créez une nouvelle session SageMaker AI à l'aide du bloc de code suivant. Cela renvoie l’ARN du rôle pour la session. L’ARN de ce rôle doit être l’ARN du rôle d’exécution que vous avez configuré comme prérequis. ``` import boto3 import sagemaker import sagemaker.session from sagemaker.workflow.pipeline_context import PipelineSession region = boto3.Session().region_name sagemaker_session = sagemaker.session.Session() role = sagemaker.get_execution_role() default_bucket = sagemaker_session.default_bucket() pipeline_session = PipelineSession() model_package_group_name = f"AbaloneModelPackageGroupName" ``` ### Création d’un pipeline **Important** Les politiques IAM personnalisées qui permettent à Amazon SageMaker Studio ou Amazon SageMaker Studio Classic de créer des SageMaker ressources Amazon doivent également accorder des autorisations pour ajouter des balises à ces ressources. L’autorisation d’ajouter des balises aux ressources est requise, car Studio et Studio Classic balisent automatiquement toutes les ressources qu’ils créent. Si une politique IAM autorise Studio et Studio Classic à créer des ressources mais n'autorise pas le balisage, des erreurs « AccessDenied » peuvent se produire lors de la tentative de création de ressources. Pour de plus amples informations, veuillez consulter [Fournir des autorisations pour le balisage des ressources d' SageMaker IA](security_iam_id-based-policy-examples.md#grant-tagging-permissions). [AWS politiques gérées pour Amazon SageMaker AI](security-iam-awsmanpol.md)qui donnent des autorisations pour créer des SageMaker ressources incluent déjà des autorisations pour ajouter des balises lors de la création de ces ressources. Exécutez les étapes suivantes à partir de votre instance de bloc-notes SageMaker AI pour créer un pipeline comprenant des étapes pour : + prétraitement + entraînement + évaluation + évaluation conditionnelle + enregistrement des modèles **Note** Vous pouvez utiliser [ExecutionVariables](https://sagemaker.readthedocs.io/en/stable/workflows/pipelines/sagemaker.workflow.pipelines.html#execution-variables)la fonction [Join](https://sagemaker.readthedocs.io/en/stable/workflows/pipelines/sagemaker.workflow.pipelines.html#execution-variables) pour spécifier votre emplacement de sortie. `ExecutionVariables`est résolu au moment de l'exécution. Par exemple, `ExecutionVariables.PIPELINE_EXECUTION_ID` est résolu avec l’ID de l’exécution en cours, qui peut être utilisé comme identifiant unique pour différentes exécutions. #### Étape 1 : Téléchargement du jeu de données Ce bloc-notes utilise le jeu de données Ormeau de machine learning de l'UCI. Le jeu de données contient les fonctions suivantes : + `length` – Mesure de la coquille la plus longue de l'ormeau. + `diameter` – Le diamètre de l'ormeau perpendiculaire à sa longueur. + `height` – La hauteur de l'ormeau avec de la viande dans la coquille. + `whole_weight` – Le poids de l'ormeau entier. + `shucked_weight` – Le poids de la viande retirée de l'ormeau. + `viscera_weight` – Poids des viscères d'ormeau après saignement. + `shell_weight` – Poids de la coquille de l'ormeau après avoir enlevé et séché la viande. + `sex` – Le sexe de l'ormeau. Une valeur « M », « F » ou « I », où « I » est un jeune ormeau. + `rings` – Le nombre d'anneaux dans la coquille de l'ormeau. Le nombre d’anneaux dans la coquille de l’ormeau est une bonne approximation de son âge en utilisant la formule `age=rings + 1.5`. Toutefois, l’obtention de ce nombre est une tâche longue. Vous devez couper la coquille à travers le cône, tacher la section et compter le nombre d’anneaux à l’aide d’un microscope. Toutefois, les autres mesures physiques sont plus faciles à obtenir. Ce bloc-notes utilise le jeu de données pour créer un modèle prédictif des anneaux variables à l’aide des autres mesures physiques. **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) ``` 1. 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) ``` #### Étape 2 : Définition des paramètres de pipeline Ce bloc de code définit les paramètres suivants pour votre pipeline : + `processing_instance_count` : le nombre d’instances de la tâche de traitement. + `input_data` : l’emplacement Amazon S3 des données d’entrée. + `batch_data` – L'emplacement Amazon S3 des données d'entrée pour la transformation par lots. + `model_approval_status` – Le statut d'approbation pour enregistrer le modèle entraîné avec pour CI/CD. Pour de plus amples informations, veuillez consulter [MLOps Automatisation avec des SageMaker projets](sagemaker-projects.md). ``` from sagemaker.workflow.parameters import ( ParameterInteger, ParameterString, ) processing_instance_count = ParameterInteger( name="ProcessingInstanceCount", default_value=1 ) model_approval_status = ParameterString( name="ModelApprovalStatus", default_value="PendingManualApproval" ) input_data = ParameterString( name="InputData", default_value=input_data_uri, ) batch_data = ParameterString( name="BatchData", default_value=batch_data_uri, ) ``` #### Étape 3 : Définition d’une étape de traitement pour l’ingénierie des caractéristiques Cette section vous indique comment créer une étape de traitement afin de préparer les données du jeu de données en vue de l'entraînement. **Pour créer une étape de traitement** 1. Créez un répertoire pour le script de traitement. ``` !mkdir -p abalone ``` 1. 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) ``` 1. 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, ) ``` 1. 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éé. Cette `run` méthode est très similaire à celle d'une instance de processeur dans le SDK SageMaker AI Python. 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 ) ``` #### Étape 4 : Définition d’une étape d’entraînement Cette section explique comment utiliser l'[XGBoostalgorithme](https://docs.aws.amazon.com/sagemaker/latest/dg/xgboost.html) d' SageMaker IA pour entraîner un modèle sur les données d'entraînement issues des étapes de traitement. **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" ``` 1. Configurez un estimateur pour l' XGBoost algorithme et le jeu de données en 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 L'IA télécharge le modèle sur Amazon S3 sous la forme d'un `model.tar.gz` document à la fin de la formation. ``` 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 ) ``` 1. 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 ) ``` #### Étape 5 : Définition d’une étape de traitement pour l’évaluation des modèles Cette section vous explique comment créer une étape de traitement pour évaluer la précision du modèle. Le résultat de cette évaluation des modèles est utilisé dans l’étape de condition pour déterminer le chemin d’exécution à prendre. **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)) ``` 1. 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, ) ``` 1. 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`. Cette `run` méthode est très similaire à celle d'une instance de processeur dans le SDK SageMaker AI Python.  ``` 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], ) ``` #### Étape 6 : Définition CreateModelStep d'une transformation par lots **Important** Nous vous recommandons [Étape du modèle](build-and-manage-steps-types.md#step-type-model) de l'utiliser pour créer des modèles à partir de la version 2.90.0 du SDK Python SageMaker . `CreateModelStep`continuera de fonctionner dans les versions précédentes du SDK SageMaker Python, mais n'est plus activement pris en charge. Cette section explique comment créer un modèle d' SageMaker IA à partir du résultat de l'étape de formation. Ce modèle est utilisé pour la transformation par lots sur un nouveau jeu de données. Cette étape est transmise à l’étape de condition et ne s’exécute que si l’étape de condition prend la valeur `true`. **Pour définir une CreateModelStep transformation par lots** 1. Créez un modèle d' SageMaker IA. 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, ) ``` 1. Définissez l'entrée du modèle pour votre modèle d' SageMaker IA. ``` from sagemaker.inputs import CreateModelInput inputs = CreateModelInput( instance_type="ml.m5.large", accelerator_type="ml.eia1.medium", ) ``` 1. Créez votre instance `CreateModelStep` à l'aide de `CreateModelInput` l'instance de modèle d' SageMaker IA que vous avez définie. ``` from sagemaker.workflow.steps import CreateModelStep step_create_model = CreateModelStep( name="AbaloneCreateModel", model=model, inputs=inputs, ) ``` #### Étape 7 : Définissez un TransformStep pour effectuer une transformation par lots Cette section explique comment créer une `TransformStep` pour effectuer une transformation par lots sur un jeu de données après l’entraînement du modèle. Cette étape est transmise à l’étape de condition et ne s’exécute que si l’étape de condition prend la valeur `true`. **Pour définir un 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" ) ``` 1. 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) ) ``` #### Étape 8 : Définition d'une RegisterModel étape pour créer un package modèle **Important** Nous vous recommandons [Étape du modèle](build-and-manage-steps-types.md#step-type-model) de l'utiliser pour enregistrer des modèles à partir de la version 2.90.0 du SDK Python SageMaker . `RegisterModel`continuera de fonctionner dans les versions précédentes du SDK SageMaker Python, mais n'est plus activement pris en charge. Cette section montre comment créer une instance de `RegisterModel`. Le résultat de l’exécution de `RegisterModel` dans un pipeline est un package de modèle. Un package de modèle est une abstraction d’artefacts de modèle réutilisable qui contient tous les ingrédients nécessaires à l’inférence. Il se compose d'une spécification d'inférence qui définit l'image d'inférence à utiliser avec un emplacement de pondération de modèle facultatif. Un groupe de packages de modèles est une collection de packages de modèles. Vous pouvez utiliser un élément `ModelPackageGroup` pour Pipelines afin d’ajouter une nouvelle version et un package de modèle au groupe pour chaque exécution de pipeline. Pour plus d’informations sur le registre de modèles, consultez [Déploiement de l’enregistrement de modèles à l’aide du registre de modèles](model-registry.md). Cette étape est transmise à l’étape de condition et ne s’exécute que si l’étape de condition prend la valeur `true`. **Pour définir une RegisterModel étape de création d'un package modèle** + 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 ) ``` #### Étape 9 : Définition d’une étape de condition pour vérifier l’exactitude du modèle Une étape `ConditionStep` permet à Pipelines de prendre en charge l’exécution conditionnelle dans le graphique DAG de votre pipeline en fonction de la condition des propriétés d’étape. Dans ce cas, vous ne souhaitez enregistrer un package de modèle que si l’exactitude de ce modèle dépasse la valeur requise. L’exactitude du modèle est déterminée par l’étape d’évaluation des modèles. Si la précision dépasse la valeur requise, le pipeline crée également un modèle d' SageMaker IA et exécute une transformation par lots sur un ensemble de données. Cette section explique comment définir l’étape Condition. **Pour définir une étape de condition pour vérifier la précision du modèle** 1. Définissez une condition `ConditionLessThanOrEqualTo`en 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 à l'aide du fichier de propriétés que vous avez indexé lors de l'étape de traitement et JSONPath de la valeur d'erreur quadratique moyenne correspondante,. `"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 ) ``` 1. 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=[], ) ``` #### Étape 10 : créer un pipeline Maintenant que vous avez créé toutes les étapes, combinez-les dans un pipeline. **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], ) ``` 1. (Facultatif) Examinez la définition de pipeline JSON pour vous assurer qu’elle est bien formée. ``` import json json.loads(pipeline.definition()) ``` Cette définition de pipeline est prête à être soumise à l' SageMaker IA. Dans le didacticiel suivant, vous soumettez ce pipeline à l' SageMaker IA et lancez une exécution. ## Définition d’un pipeline (JSON) Vous pouvez également utiliser [boto3](https://boto3.amazonaws.com/v1/documentation/api/latest/reference/services/sagemaker.html#SageMaker.Client.create_pipeline) ou [CloudFormation](https://docs.aws.amazon.com/AWSCloudFormation/latest/UserGuide/aws-resource-sagemaker-pipeline.html) pour créer un pipeline. La création d’un pipeline nécessite une définition de pipeline, qui est un objet JSON définissant chaque étape du pipeline. Le SageMaker SDK offre un moyen simple de créer la définition du pipeline, que vous pouvez utiliser avec n'importe lequel des éléments mentionnés APIs précédemment pour créer le pipeline lui-même. Sans utiliser le SDK, les utilisateurs doivent écrire la définition JSON brute pour créer le pipeline sans aucune des vérifications d'erreur fournies par le SDK SageMaker Python. Pour voir le schéma de la définition JSON du pipeline, consultez le [schéma JSON de définition du pipeline SageMaker AI](https://aws-sagemaker-mlops.github.io/sagemaker-model-building-pipeline-definition-JSON-schema/). L'exemple de code suivant montre un exemple d'objet JSON de définition de pipeline d' SageMaker IA : ``` {'Version': '2020-12-01', 'Metadata': {}, 'Parameters': [{'Name': 'ProcessingInstanceType', 'Type': 'String', 'DefaultValue': 'ml.m5.xlarge'}, {'Name': 'ProcessingInstanceCount', 'Type': 'Integer', 'DefaultValue': 1}, {'Name': 'TrainingInstanceType', 'Type': 'String', 'DefaultValue': 'ml.m5.xlarge'}, {'Name': 'ModelApprovalStatus', 'Type': 'String', 'DefaultValue': 'PendingManualApproval'}, {'Name': 'ProcessedData', 'Type': 'String', 'DefaultValue': 'S3_URL', {'Name': 'InputDataUrl', 'Type': 'String', 'DefaultValue': 'S3_URL', 'PipelineExperimentConfig': {'ExperimentName': {'Get': 'Execution.PipelineName'}, 'TrialName': {'Get': 'Execution.PipelineExecutionId'}}, 'Steps': [{'Name': 'ReadTrainDataFromFS', 'Type': 'Processing', 'Arguments': {'ProcessingResources': {'ClusterConfig': {'InstanceType': 'ml.m5.4xlarge', 'InstanceCount': 2, 'VolumeSizeInGB': 30}}, 'AppSpecification': {'ImageUri': 'IMAGE_URI', 'ContainerArguments': [....]}, 'RoleArn': 'ROLE', 'ProcessingInputs': [...], 'ProcessingOutputConfig': {'Outputs': [.....]}, 'StoppingCondition': {'MaxRuntimeInSeconds': 86400}}, 'CacheConfig': {'Enabled': True, 'ExpireAfter': '30d'}}, ... ... ... } ``` **Étape suivante :** [Exécuter un pipeline](run-pipeline.md)