Réduisez les coûts et le temps de développement avec le mode local d'Amazon SageMaker Pipelines

La création de pipelines d'apprentissage automatique (ML) robustes et réutilisables peut être un processus complexe et chronophage. Les développeurs testent généralement leurs scripts de traitement et de formation localement, mais les pipelines eux-mêmes sont généralement testés dans le cloud. La création et l'exécution d'un pipeline complet pendant l'expérimentation ajoutent des frais généraux et des coûts indésirables au cycle de vie du développement. Dans cet article, nous détaillons comment vous pouvez utiliser Mode local Amazon SageMaker Pipelines pour exécuter des pipelines ML localement afin de réduire à la fois le développement du pipeline et le temps d'exécution tout en réduisant les coûts. Une fois le pipeline entièrement testé localement, vous pouvez facilement le réexécuter avec Amazon Sage Maker ressources gérées avec seulement quelques lignes de modifications de code.

Présentation du cycle de vie du ML

L'un des principaux moteurs des nouvelles innovations et applications en ML est la disponibilité et la quantité de données, ainsi que des options de calcul moins chères. Dans plusieurs domaines, le ML s'est avéré capable de résoudre des problèmes auparavant insolubles avec les mégadonnées classiques et les techniques analytiques, et la demande de spécialistes de la science des données et du ML augmente régulièrement. À un niveau très élevé, le cycle de vie ML se compose de nombreuses parties différentes, mais la construction d'un modèle ML comprend généralement les étapes générales suivantes :

1. Nettoyage et préparation des données (feature engineering)
2. Formation et réglage du modèle
3. Évaluation du modèle
4. Déploiement de modèle (ou transformation par lots)

Lors de l'étape de préparation des données, les données sont chargées, massées et transformées en type d'entrées ou fonctionnalités attendues par le modèle ML. L'écriture des scripts pour transformer les données est généralement un processus itératif, où des boucles de rétroaction rapides sont importantes pour accélérer le développement. Il n'est normalement pas nécessaire d'utiliser l'ensemble de données complet lors du test de scripts d'ingénierie de fonctionnalités, c'est pourquoi vous pouvez utiliser le fonctionnalité de mode local du traitement SageMaker. Cela vous permet d'exécuter localement et de mettre à jour le code de manière itérative, en utilisant un jeu de données plus petit. Lorsque le code final est prêt, il est soumis au travail de traitement à distance, qui utilise l'ensemble de données complet et s'exécute sur des instances gérées par SageMaker.

Le processus de développement est similaire à l'étape de préparation des données pour les étapes de formation et d'évaluation du modèle. Les data scientists utilisent fonctionnalité de mode local de SageMaker Training pour itérer rapidement avec des ensembles de données plus petits localement, avant d'utiliser toutes les données dans un cluster géré par SageMaker d'instances optimisées pour le ML. Cela accélère le processus de développement et élimine le coût d'exécution des instances ML gérées par SageMaker pendant l'expérimentation.

À mesure que la maturité ML d'une organisation augmente, vous pouvez utiliser Pipelines Amazon SageMaker pour créer des pipelines ML qui assemblent ces étapes, créant des flux de travail ML plus complexes qui traitent, forment et évaluent des modèles ML. SageMaker Pipelines est un service entièrement géré pour automatiser les différentes étapes du flux de travail ML, y compris le chargement des données, la transformation des données, la formation et le réglage du modèle et le déploiement du modèle. Jusqu'à récemment, vous pouviez développer et tester vos scripts localement, mais vous deviez tester vos pipelines ML dans le cloud. Cela a rendu l'itération sur le flux et la forme des pipelines ML un processus lent et coûteux. Désormais, avec la fonctionnalité de mode local ajoutée de SageMaker Pipelines, vous pouvez itérer et tester vos pipelines ML de la même manière que vous testez et itérez sur vos scripts de traitement et de formation. Vous pouvez exécuter et tester vos pipelines sur votre machine locale, en utilisant un petit sous-ensemble de données pour valider la syntaxe et les fonctionnalités du pipeline.

Pipelines SageMaker

SageMaker Pipelines offre un moyen entièrement automatisé d'exécuter des workflows ML simples ou complexes. Avec SageMaker Pipelines, vous pouvez créer des flux de travail ML avec un SDK Python facile à utiliser, puis visualiser et gérer votre flux de travail à l'aide Amazon SageMakerStudio. Vos équipes de science des données peuvent être plus efficaces et évoluer plus rapidement en stockant et en réutilisant les étapes de workflow que vous créez dans SageMaker Pipelines. Vous pouvez également utiliser des modèles prédéfinis qui automatisent la création de l'infrastructure et du référentiel pour créer, tester, enregistrer et déployer des modèles dans votre environnement ML. Ces modèles sont automatiquement disponibles pour votre organisation et sont provisionnés à l'aide de Catalogue de services AWS en vente au détail.

SageMaker Pipelines apporte des pratiques d'intégration et de déploiement continus (CI/CD) au ML, telles que le maintien de la parité entre les environnements de développement et de production, le contrôle de version, les tests à la demande et l'automatisation de bout en bout, ce qui vous aide à faire évoluer le ML tout au long de votre organisme. Les praticiens DevOps savent que certains des principaux avantages de l'utilisation des techniques CI/CD incluent une augmentation de la productivité via des composants réutilisables et une augmentation de la qualité grâce à des tests automatisés, ce qui conduit à un retour sur investissement plus rapide pour vos objectifs commerciaux. Ces avantages sont désormais disponibles pour les praticiens MLOps en utilisant SageMaker Pipelines pour automatiser la formation, les tests et le déploiement des modèles ML. Avec le mode local, vous pouvez désormais itérer beaucoup plus rapidement lors du développement de scripts à utiliser dans un pipeline. Notez que les instances de pipeline locales ne peuvent pas être affichées ou exécutées dans l'IDE Studio ; cependant, des options de visualisation supplémentaires pour les pipelines locaux seront bientôt disponibles.

Le SDK SageMaker fournit un objectif général configuration du mode local qui permet aux développeurs d'exécuter et de tester les processeurs et les estimateurs pris en charge dans leur environnement local. Vous pouvez utiliser la formation en mode local avec plusieurs images de structure prises en charge par AWS (TensorFlow, MXNet, Chainer, PyTorch et Scikit-Learn) ainsi que des images que vous fournissez vous-même.

SageMaker Pipelines, qui crée un graphique acyclique dirigé (DAG) d'étapes de workflow orchestrées, prend en charge de nombreuses activités qui font partie du cycle de vie ML. En mode local, les étapes suivantes sont prises en charge :

• Traitement des étapes de travail – Une expérience simplifiée et gérée sur SageMaker pour exécuter des charges de travail de traitement de données, telles que l'ingénierie des fonctionnalités, la validation des données, l'évaluation des modèles et l'interprétation des modèles
• Étapes de la tâche de formation – Un processus itératif qui apprend à un modèle à faire des prédictions en présentant des exemples à partir d'un ensemble de données de formation
• Tâches de réglage d'hyperparamètres – Un moyen automatisé d'évaluer et de sélectionner les hyperparamètres qui produisent le modèle le plus précis
• Étapes d'exécution conditionnelles – Une étape qui fournit une série conditionnelle de branches dans un pipeline
• Étape du modèle – À l'aide des arguments CreateModel, cette étape peut créer un modèle à utiliser dans les étapes de transformation ou un déploiement ultérieur en tant que point de terminaison
• Transformer les étapes du travail – Une tâche de transformation par lots qui génère des prédictions à partir de grands ensembles de données et exécute l'inférence lorsqu'un point de terminaison persistant n'est pas nécessaire
• Échec des étapes – Une étape qui arrête l'exécution d'un pipeline et marque l'exécution comme ayant échoué

Vue d'ensemble de la solution

Notre solution illustre les étapes essentielles pour créer et exécuter SageMaker Pipelines en mode local, ce qui signifie utiliser des ressources locales de CPU, de RAM et de disque pour charger et exécuter les étapes du flux de travail. Votre environnement local peut être exécuté sur un ordinateur portable, en utilisant des IDE populaires tels que VSCode ou PyCharm, ou il peut être hébergé par SageMaker à l'aide d'instances de notebook classiques.

Le mode local permet aux scientifiques des données d'assembler des étapes, qui peuvent inclure des tâches de traitement, de formation et d'évaluation, et d'exécuter l'intégralité du flux de travail localement. Lorsque vous avez terminé les tests locaux, vous pouvez réexécuter le pipeline dans un environnement géré par SageMaker en remplaçant le LocalPipelineSession objet avec PipelineSession, ce qui apporte de la cohérence au cycle de vie du ML.

Pour cet exemple de bloc-notes, nous utilisons un ensemble de données standard accessible au public, le Ensemble de données d'ormeau d'apprentissage automatique UCI. L'objectif est de former un modèle ML pour déterminer l'âge d'un escargot d'ormeau à partir de ses mesures physiques. Au fond, c'est un problème de régression.

Tout le code requis pour exécuter cet exemple de bloc-notes est disponible sur GitHub dans le amazon-sagemaker-exemples dépôt. Dans cet exemple de bloc-notes, chaque étape de workflow de pipeline est créée indépendamment, puis câblée ensemble pour créer le pipeline. Nous créons les étapes suivantes :

• Étape de traitement (ingénierie des fonctionnalités)
• Étape d'entraînement (entraînement du modèle)
• Étape de traitement (évaluation du modèle)
• Étape de condition (précision du modèle)
• Créer une étape de modèle (modèle)
• Étape de transformation (transformation par lots)
• Enregistrer l'étape du modèle (paquet de modèles)
• Échec de l'étape (l'exécution a échoué)

Le schéma suivant illustre notre pipeline.

Pré-requis

Pour suivre cet article, vous avez besoin des éléments suivants :

Une fois ces conditions préalables en place, vous pouvez exécuter l'exemple de bloc-notes comme décrit dans les sections suivantes.

Construisez votre pipeline

Dans cet exemple de notebook, nous utilisons Mode de script SageMaker pour la plupart des processus ML, ce qui signifie que nous fournissons le code Python réel (scripts) pour effectuer l'activité et transmettons une référence à ce code. Le mode Script offre une grande flexibilité pour contrôler le comportement dans le traitement SageMaker en vous permettant de personnaliser votre code tout en profitant des conteneurs prédéfinis SageMaker tels que XGBoost ou Scikit-Learn. Le code personnalisé est écrit dans un fichier de script Python à l'aide de cellules commençant par la commande magique %%writefile, comme ce qui suit :

%%writefile code/evaluation.py

Le principal catalyseur du mode local est le LocalPipelineSession objet, qui est instancié à partir du SDK Python. Les segments de code suivants montrent comment créer un pipeline SageMaker en mode local. Bien que vous puissiez configurer un chemin de données local pour de nombreuses étapes de pipeline locales, Amazon S3 est l'emplacement par défaut pour stocker les données générées par la transformation. Le nouveau LocalPipelineSession L'objet est transmis au SDK Python dans de nombreux appels d'API de workflow SageMaker décrits dans cet article. Notez que vous pouvez utiliser le local_pipeline_session variable pour récupérer les références au compartiment par défaut S3 et au nom de la région actuelle.

from sagemaker.workflow.pipeline_context import LocalPipelineSession

# Create a `LocalPipelineSession` object so that each 
# pipeline step will run locally
# To run this pipeline in the cloud, you must change 
# the `LocalPipelineSession()` to `PipelineSession()`
local_pipeline_session = LocalPipelineSession()
region = local_pipeline_session.boto_region_name

default_bucket = local_pipeline_session.default_bucket()
prefix = "sagemaker-pipelines-local-mode-example"

Avant de créer les étapes de pipeline individuelles, nous définissons certains paramètres utilisés par le pipeline. Certains de ces paramètres sont des littéraux de chaîne, tandis que d'autres sont créés en tant que types énumérés spéciaux fournis par le SDK. Le typage énuméré garantit que des paramètres valides sont fournis au pipeline, comme celui-ci, qui est transmis au ConditionLessThanOrEqualTo descendre plus bas :

mse_threshold = ParameterFloat(name="MseThreshold", default_value=7.0)

Pour créer une étape de traitement de données, qui est utilisée ici pour effectuer l'ingénierie des fonctionnalités, nous utilisons le SKLearnProcessor pour charger et transformer le jeu de données. Nous passons le local_pipeline_session variable au constructeur de classe, qui indique à l'étape du workflow de s'exécuter en mode local :

from sagemaker.sklearn.processing import SKLearnProcessor

framework_version = "1.0-1"

sklearn_processor = SKLearnProcessor(
    framework_version=framework_version,
    instance_type=instance_type,
    instance_count=processing_instance_count,
    base_job_name="sklearn-abalone-process",
    role=role,
    sagemaker_session=local_pipeline_session,
)

Ensuite, nous créons notre première étape de pipeline réelle, un ProcessingStep objet, tel qu'importé depuis le SDK SageMaker. Les arguments du processeur sont retournés à partir d'un appel au SKLearnProcessor méthode run(). Cette étape du flux de travail est combinée avec d'autres étapes vers la fin du bloc-notes pour indiquer l'ordre des opérations dans le pipeline.

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="code/preprocessing.py",
)

step_process = ProcessingStep(name="AbaloneProcess", step_args=processor_args)

Ensuite, nous fournissons du code pour établir une étape de formation en instanciant d'abord un estimateur standard à l'aide du SDK SageMaker. Nous passons le même local_pipeline_session variable à l'estimateur, nommé xgb_train, comme la sagemaker_session dispute. Parce que nous voulons former un modèle XGBoost, nous devons générer un URI d'image valide en spécifiant les paramètres suivants, y compris le framework et plusieurs paramètres de version :

from sagemaker.estimator import Estimator
from sagemaker.inputs import TrainingInput

model_path = f"s3://{default_bucket}/{prefix}/model"
image_uri = sagemaker.image_uris.retrieve(
    framework="xgboost",
    region=region,
    version="1.5-1",
    py_version="py3",
    instance_type=instance_type,
)

xgb_train = Estimator(
    image_uri=image_uri,
    entry_point="code/abalone.py",
    instance_type=instance_type,
    instance_count=training_instance_count,
    output_path=model_path,
    role=role,
    sagemaker_session=local_pipeline_session,
)

Nous pouvons éventuellement appeler des méthodes d'estimation supplémentaires, par exemple set_hyperparameters(), afin de fournir des paramètres d'hyperparamètres pour la tâche d'entraînement. Maintenant que nous avons configuré un estimateur, nous sommes prêts à créer l'étape de formation proprement dite. Encore une fois, nous importons le TrainingStep classe de la bibliothèque SageMaker SDK :

from sagemaker.workflow.steps import TrainingStep

step_train = TrainingStep(name="AbaloneTrain", step_args=train_args)

Ensuite, nous construisons une autre étape de traitement pour effectuer l'évaluation du modèle. Cela se fait en créant un ScriptProcessor instance et en passant le local_pipeline_session objet en paramètre :

from sagemaker.processing import ScriptProcessor

script_eval = ScriptProcessor(
    image_uri=image_uri,
    command=["python3"],
    instance_type=instance_type,
    instance_count=processing_instance_count,
    base_job_name="script-abalone-eval",
    role=role,
    sagemaker_session=local_pipeline_session,
)

Pour permettre le déploiement du modèle entraîné, soit vers un Point de terminaison en temps réel SageMaker ou à une transformation par lots, nous devons créer un Model objet en transmettant les artefacts du modèle, l'URI de l'image appropriée et éventuellement notre code d'inférence personnalisé. Nous passons ensuite ce Model s'opposer à un ModelStep, qui est ajouté au pipeline local. Voir le code suivant :

from sagemaker.model import Model

model = Model(
    image_uri=image_uri,
    model_data=step_train.properties.ModelArtifacts.S3ModelArtifacts,
    source_dir="code",
    entry_point="inference.py",
    role=role,
    sagemaker_session=local_pipeline_session,
)

from sagemaker.workflow.model_step import ModelStep

step_create_model = ModelStep(name="AbaloneCreateModel", 
    step_args=model.create(instance_type=instance_type)
)

Ensuite, nous créons une étape de transformation par lots où nous soumettons un ensemble de vecteurs de caractéristiques et effectuons une inférence. Nous devons d'abord créer un Transformer objet et passer le local_pipeline_session paramètre à elle. Ensuite, nous créons un TransformStep, en transmettant les arguments requis, et ajoutez ceci à la définition du pipeline :

from sagemaker.transformer import Transformer

transformer = Transformer(
    model_name=step_create_model.properties.ModelName,
    instance_type=instance_type,
    instance_count=transform_instance_count,
    output_path=f"s3://{default_bucket}/{prefix}/transform",
    sagemaker_session=local_pipeline_session,
)

from sagemaker.workflow.steps import TransformStep

transform_args = transformer.transform(transform_data, content_type="text/csv")

step_transform = TransformStep(name="AbaloneTransform", step_args=transform_args)

Enfin, nous souhaitons ajouter une condition de branche au flux de travail afin de n'exécuter la transformation par lots que si les résultats de l'évaluation du modèle répondent à nos critères. Nous pouvons indiquer ce conditionnel en ajoutant un ConditionStep avec un type de condition particulier, comme ConditionLessThanOrEqualTo. Nous énumérons ensuite les étapes pour les deux branches, définissant essentiellement les branches if/else ou true/false du pipeline. Les if_steps fournis dans le ConditionStep (step_create_model, étape_transformation) sont exécutés chaque fois que la condition est évaluée à True.

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=mse_threshold,
)

step_cond = ConditionStep(
    name="AbaloneMSECond",
    conditions=[cond_lte],
    if_steps=[step_create_model, step_transform],
    else_steps=[step_fail],
)

Le schéma suivant illustre cette branche conditionnelle et les étapes if/else associées. Une seule branche est exécutée, basée sur le résultat de l'étape d'évaluation du modèle par rapport à l'étape de condition.

Maintenant que toutes nos étapes sont définies et que les instances de classe sous-jacentes sont créées, nous pouvons les combiner dans un pipeline. Nous fournissons certains paramètres et définissons de manière cruciale l'ordre des opérations en énumérant simplement les étapes dans l'ordre souhaité. Notez que le TransformStep n'est pas affiché ici car il s'agit de la cible de l'étape conditionnelle et a été fourni comme argument d'étape à la ConditionalStep plus tôt.

from sagemaker.workflow.pipeline import Pipeline

pipeline_name = f"LocalModelPipeline"
pipeline = Pipeline(
    name=pipeline_name,
    parameters=[
        input_data,
        mse_threshold,
    ],
    steps=[step_process, step_train, step_eval, step_cond],
    sagemaker_session=local_pipeline_session,
)

Pour exécuter le pipeline, vous devez appeler deux méthodes : pipeline.upsert(), qui télécharge le pipeline vers le service sous-jacent, et pipeline.start(), qui démarre l'exécution du pipeline. Vous pouvez utiliser diverses autres méthodes pour interroger l'état d'exécution, répertorier les étapes du pipeline, etc. Étant donné que nous avons utilisé la session de pipeline en mode local, ces étapes sont toutes exécutées localement sur votre processeur. La sortie de la cellule sous la méthode start montre la sortie du pipeline :

pipeline.upsert(role_arn=role)
execution = pipeline.start()

Vous devriez voir un message au bas de la sortie de la cellule semblable au suivant :

Pipeline execution d8c3e172-089e-4e7a-ad6d-6d76caf987b7 SUCCEEDED

Revenir aux ressources gérées

Une fois que nous avons confirmé que le pipeline s'exécute sans erreur et que nous sommes satisfaits du flux et de la forme du pipeline, nous pouvons recréer le pipeline mais avec des ressources gérées par SageMaker et le réexécuter. Le seul changement requis est d'utiliser le PipelineSession objet au lieu de LocalPipelineSession:

de sagemaker.workflow.pipeline_context importer LocalPipelineSession
from sagemaker.workflow.pipeline_context import PipelineSession

local_pipeline_session=LocalPipelineSession()
pipeline_session = PipelineSession()

Cela informe le service d'exécuter chaque étape faisant référence à cet objet de session sur les ressources gérées SageMaker. Compte tenu du petit changement, nous illustrons uniquement les modifications de code requises dans la cellule de code suivante, mais la même modification devrait être implémentée sur chaque cellule à l'aide de la local_pipeline_session objet. Les modifications sont cependant identiques dans toutes les cellules, car nous ne remplaçons que le local_pipeline_session objet avec le pipeline_session objet.

from sagemaker.sklearn.processing import SKLearnProcessor

framework_version = "1.0-1"

sklearn_processor = SKLearnProcessor(
    framework_version=framework_version,
    instance_type=instance_type,
    instance_count=processing_instance_count,
    base_job_name="sklearn-abalone-process",
    role=role,
    sagemaker_session=pipeline_session,  # non-local session
)

Une fois l'objet de session local remplacé partout, nous recréons le pipeline et l'exécutons avec les ressources gérées SageMaker :

from sagemaker.workflow.pipeline import Pipeline

pipeline_name = f"LocalModelPipeline"
pipeline = Pipeline(
    name=pipeline_name,
    parameters=[
        input_data,
        mse_threshold,
    ],
    steps=[step_process, step_train, step_eval, step_cond],
    sagemaker_session=pipeline_session, # non-local session
)

pipeline.upsert(role_arn=role)
execution = pipeline.start()

Nettoyer

Si vous souhaitez conserver l'environnement Studio propre, vous pouvez utiliser les méthodes suivantes pour supprimer le pipeline SageMaker et le modèle. Le code complet peut être trouvé dans l'exemple cahier.

# delete models 
sm_client = boto3.client("sagemaker")
model_prefix="AbaloneCreateModel"
delete_models(sm_client, model_prefix)

# delete managed pipeline
pipeline_to_delete = 'SM-Managed-Pipeline'
delete_sagemaker_pipeline(sm_client, pipeline_to_delete)

Conclusion

Jusqu'à récemment, vous pouviez utiliser la fonction de mode local de SageMaker Processing et SageMaker Training pour itérer localement sur vos scripts de traitement et de formation, avant de les exécuter sur toutes les données avec les ressources gérées SageMaker. Avec la nouvelle fonctionnalité de mode local de SageMaker Pipelines, les praticiens ML peuvent désormais appliquer la même méthode lors de l'itération sur leurs pipelines ML, en assemblant les différents flux de travail ML ensemble. Lorsque le pipeline est prêt pour la production, son exécution avec les ressources gérées SageMaker ne nécessite que quelques lignes de modifications de code. Cela réduit le temps d'exécution du pipeline pendant le développement, ce qui permet un développement plus rapide du pipeline avec des cycles de développement plus rapides, tout en réduisant le coût des ressources gérées par SageMaker.

Pour en savoir plus, visitez le site Pipelines Amazon SageMaker or Utilisez les pipelines SageMaker pour exécuter vos tâches localement.

À propos des auteurs

Paul Hargis a concentré ses efforts sur l'apprentissage automatique dans plusieurs entreprises, dont AWS, Amazon et Hortonworks. Il aime créer des solutions technologiques et enseigner aux gens comment en tirer le meilleur parti. Avant d'occuper son poste chez AWS, il était architecte principal pour les exportations et les extensions d'Amazon, aidant amazon.com à améliorer l'expérience des acheteurs internationaux. Paul aime aider les clients à développer leurs initiatives d'apprentissage automatique pour résoudre des problèmes réels.

Nicolas Palme est un architecte de solutions chez AWS à Stockholm, en Suède, où il aide les clients des pays nordiques à réussir dans le cloud. Il est particulièrement passionné par les technologies sans serveur, ainsi que par l'IoT et l'apprentissage automatique. En dehors du travail, Niklas est un passionné de ski de fond et de snowboard ainsi qu'un maître bouilleur d'œufs.

Kirit Thadaka est un architecte de solutions ML travaillant dans l'équipe SageMaker Service SA. Avant de rejoindre AWS, Kirit a travaillé dans des startups d'IA à un stade précoce, puis a été consultant dans divers rôles dans la recherche sur l'IA, les MLOps et la direction technique.