Intégrez Amazon SageMaker Data Wrangler aux workflows MLOps

Alors que les entreprises passent de l'exécution de modèles d'apprentissage automatique (ML) ad hoc à l'utilisation de l'IA/ML pour transformer leur activité à grande échelle, l'adoption des opérations ML (MLOps) devient inévitable. Comme le montre la figure suivante, le cycle de vie ML commence par définir un problème métier comme un cas d'utilisation ML suivi d'une série de phases, y compris la préparation des données, l'ingénierie des fonctionnalités, la création de modèles, le déploiement, la surveillance continue et le recyclage. Pour de nombreuses entreprises, bon nombre de ces étapes sont encore manuelles et vaguement intégrées les unes aux autres. Par conséquent, il est important d'automatiser le cycle de vie ML de bout en bout, ce qui permet des expériences fréquentes pour générer de meilleurs résultats commerciaux. La préparation des données est l'une des étapes cruciales de ce cycle de vie, car la précision du modèle ML dépend de la qualité de l'ensemble de données d'apprentissage.

Les data scientists et les ingénieurs ML passent entre 70 et 80 % de leur temps à collecter, analyser, nettoyer et transformer les données nécessaires à la formation des modèles. Gestionnaire de données Amazon SageMaker est une capacité entièrement gérée de Amazon Sage Maker cela permet aux data scientists et aux ingénieurs ML d'analyser et de préparer plus rapidement les données pour leurs projets ML avec peu ou pas de code. Lorsqu'il s'agit d'opérationnaliser un cycle de vie ML de bout en bout, la préparation des données est presque toujours la première étape du processus. Étant donné qu'il existe de nombreuses façons de créer un pipeline ML de bout en bout, dans cet article, nous expliquons comment vous pouvez facilement intégrer Data Wrangler à certaines des technologies d'automatisation et d'orchestration des flux de travail bien connues.

Vue d'ensemble de la solution

Dans cet article, nous montrons comment les utilisateurs peuvent intégrer la préparation des données à l'aide de Data Wrangler avec Pipelines Amazon SageMaker, Fonctions d'étape AWSet une Flux d'air Apache comprenant Flux de travail géré par Amazon pour Apache Airflow (Amazon MWAA). Pipelines est une fonctionnalité de SageMaker qui est un service d'intégration continue et de livraison continue (CI/CD) spécialement conçu et facile à utiliser pour le ML. Step Functions est un service de workflow visuel low-code sans serveur utilisé pour orchestrer les services AWS et automatiser les processus métier. Amazon MWAA est un service d'orchestration géré pour Apache Airflow qui facilite l'exploitation des données de bout en bout et des pipelines ML.

À des fins de démonstration, nous considérons un cas d'utilisation pour préparer des données pour former un modèle ML avec le SageMaker intégré Algorithme XGBoost qui nous aideront à identifier les réclamations d'assurance véhicule frauduleuses. Nous avons utilisé un ensemble d'échantillons de données générés de manière synthétique pour entraîner le modèle et créer un modèle SageMaker à l'aide des artefacts de modèle issus du processus d'entraînement. Notre objectif est d'opérationnaliser ce processus de bout en bout en mettant en place un workflow ML. Bien que les flux de travail ML puissent être plus élaborés, nous utilisons un flux de travail minimal à des fins de démonstration. La première étape du workflow est la préparation des données avec Data Wrangler, suivie d'une étape de formation du modèle, et enfin d'une étape de création du modèle. Le diagramme suivant illustre le flux de travail de notre solution.

Dans les sections suivantes, nous vous expliquons comment configurer un flux Data Wrangler et intégrer Data Wrangler avec Pipelines, Step Functions et Apache Airflow.

Configurer un flux Data Wrangler

Nous commençons par créer un flux Data Wrangler, également appelé flux flux de données, En utilisant l' interface utilisateur de flux de données via le Amazon SageMakerStudio IDE. Notre exemple de jeu de données se compose de deux fichiers de données : revendications.csv ainsi que clients.csv, qui sont stockés dans un Service de stockage simple Amazon (Amazon S3) seau. Nous utilisons l'interface utilisateur du flux de données pour appliquer les règles de Data Wrangler transformations intégrées tel encodage catégoriel, formatage de chaîne et imputation aux colonnes de caractéristiques dans chacun de ces fichiers. Nous appliquons également transformation personnalisée à quelques colonnes de fonctionnalités en utilisant quelques lignes de code Python personnalisé avec Cadre de données Pandas. La capture d'écran suivante montre les transformations appliquées au fichier claims.csv dans l'interface utilisateur du flux de données.

Enfin, nous joignons les résultats des transformations appliquées des deux fichiers de données pour générer un seul ensemble de données d'entraînement pour notre entraînement de modèle. Nous utilisons la fonction intégrée de Data Wrangler joindre des ensembles de données capacité, qui nous permet d'effectuer des opérations de jointure de type SQL sur des données tabulaires. La capture d'écran suivante montre le flux de données dans l'interface utilisateur du flux de données dans Studio. Pour obtenir des instructions détaillées sur la création du flux de données à l'aide de Data Wrangler, reportez-vous au GitHub référentiel.

Vous pouvez maintenant utiliser le fichier de flux de données (.flow) pour effectuer des transformations de données sur nos fichiers de données brutes. L'interface utilisateur du flux de données peut générer automatiquement des blocs-notes Python que nous pouvons utiliser et intégrer directement aux Pipelines à l'aide du SDK SageMaker. Pour les fonctions pas à pas, nous utilisons les AWS Step Functions SDK Python de science des données pour intégrer notre traitement Data Wrangler avec un pipeline Step Functions. Pour Amazon MWAA, nous utilisons un opérateur de flux d'air personnalisé et par Opérateur SageMaker. Nous abordons chacune de ces approches en détail dans les sections suivantes.

Intégrer Data Wrangler aux pipelines

SageMaker Pipelines est un outil d'orchestration de flux de travail natif pour la création de pipelines ML qui tirent parti de l'intégration directe de SageMaker. Avec le Registre de modèles SageMaker, Pipelines améliore la résilience opérationnelle et la reproductibilité de vos workflows de ML. Ces composants d'automatisation des flux de travail vous permettent d'adapter facilement votre capacité à créer, former, tester et déployer des centaines de modèles en production ; itérer plus vite ; réduire les erreurs dues à l'orchestration manuelle ; et construire des mécanismes reproductibles. Chaque étape du pipeline peut suivre la lignée et les étapes intermédiaires peuvent être mises en cache pour réexécuter rapidement le pipeline. Tu peux créer des pipelines à l'aide du SDK SageMaker Python.

Un flux de travail construit avec les pipelines SageMaker consiste en une séquence d'étapes formant un graphe acyclique dirigé (DAG). Dans cet exemple, nous commençons par un étape de traitement, qui gère un Travail de traitement SageMaker basé sur le fichier de flux de Data Wrangler pour créer un ensemble de données de formation. On continue ensuite avec un étape de formation, où nous formons un modèle XGBoost à l'aide de l'algorithme XGBoost intégré de SageMaker et de l'ensemble de données de formation créé à l'étape précédente. Une fois qu'un modèle a été formé, nous terminons ce flux de travail avec un Étape RegisterModel pour enregistrer le modèle formé auprès du registre de modèles SageMaker.

Installation et procédure pas à pas

Pour exécuter cet exemple, nous utilisons un notebook Jupyter exécutant Python3 sur une image de noyau Data Science dans un environnement Studio. Vous pouvez également l'exécuter sur une instance de bloc-notes Jupyter localement sur votre machine en configurant les informations d'identification pour assumer le rôle d'exécution SageMaker. Le bloc-notes est léger et peut être exécuté sur une instance ml.t3.medium. Des instructions détaillées étape par étape se trouvent dans le GitHub référentiel.

Vous pouvez soit utiliser la fonction d'exportation dans Data Wrangler pour générer le code Pipelines, soit créer votre propre script à partir de zéro. Dans notre exemple de référentiel, nous utilisons une combinaison des deux approches pour plus de simplicité. De manière générale, les étapes suivantes permettent de créer et d'exécuter le flux de travail Pipelines :

1. Générez un fichier de flux à partir de Data Wrangler ou utilisez le script de configuration pour générer un fichier de flux à partir d'un modèle préconfiguré.
2. Créer un Service de stockage simple Amazon (Amazon S3) et chargez votre fichier de flux et vos fichiers d'entrée dans le compartiment. Dans notre exemple de bloc-notes, nous utilisons le compartiment S3 par défaut de SageMaker.
3. Suivez les instructions du bloc-notes pour créer un objet Processor basé sur le fichier de flux Data Wrangler et un objet Estimator avec les paramètres de la tâche d'entraînement.
   1. Dans notre exemple, comme nous n'utilisons que les fonctionnalités de SageMaker et le compartiment S3 par défaut, nous pouvons utiliser le rôle d'exécution par défaut de Studio. Le même Gestion des identités et des accès AWS (IAM) est assumé par l'exécution du pipeline, la tâche de traitement et la tâche de formation. Vous pouvez personnaliser davantage le rôle d'exécution en fonction du privilège minimum.
4. Continuez avec les instructions pour créer un pipeline avec des étapes faisant référence aux objets Processor et Estimator, puis exécutez le pipeline. Les tâches de traitement et de formation s'exécutent sur des environnements gérés par SageMaker et prennent quelques minutes.
5. Dans Studio, vous pouvez voir les détails du pipeline surveiller l'exécution du pipeline. Vous pouvez également surveiller les tâches de traitement et de formation sous-jacentes à partir de la console SageMaker ou de Amazon Cloud Watch.

Intégrer Data Wrangler avec Step Functions

Avec Step Functions, vous pouvez exprimer une logique métier complexe sous forme de flux de travail pilotés par des événements et à faible code qui connectent différents services AWS. La Fonctions d'étape SDK de science des données est une bibliothèque open source qui permet aux scientifiques des données de créer des flux de travail capables de prétraiter des ensembles de données et de créer, déployer et surveiller des modèles ML à l'aide de SageMaker et de Step Functions. Step Functions est basé sur des machines d'état et des tâches. Step Functions crée des workflows à partir d'étapes appelées Etats, et exprime ce flux de travail dans le Langue Amazon States. Lorsque vous créez un flux de travail à l'aide du SDK Step Functions Data Science, il crée une machine d'état représentant votre flux de travail et vos étapes dans Step Functions.

Pour ce cas d'utilisation, nous avons construit un flux de travail Step Functions basé sur le modèle commun utilisé dans cet article qui comprend une étape de traitement, une étape de formation et RegisterModel marcher. Dans ce cas, nous importons ces étapes à partir du SDK Step Functions Data Science Python. Nous enchaînons ces étapes dans le même ordre pour créer un workflow Step Functions. Le workflow utilise le fichier de flux généré à partir de Data Wrangler, mais vous pouvez également utiliser votre propre fichier de flux Data Wrangler. Nous réutilisons une partie du code du Data Wrangler fonction d'exportation pour la simplicité. Nous exécutons la logique de prétraitement des données générée par le fichier de flux Data Wrangler pour créer un ensemble de données de formation, former un modèle à l'aide de l'algorithme XGBoost et enregistrer l'artefact de modèle formé en tant que modèle SageMaker. De plus, dans le référentiel GitHub, nous montrons également comment Step Functions nous permet d'essayer de détecter les erreurs, et de gérer les échecs et les tentatives avec FailStateStep ainsi que CatchStateStep.

Le diagramme de flux résultant, comme illustré dans la capture d'écran suivante, est disponible sur la console Step Functions après le démarrage du workflow. Cela aide les data scientists et les ingénieurs à visualiser l'ensemble du flux de travail et chaque étape de celui-ci, et à accéder aux journaux CloudWatch liés pour chaque étape.

Installation et procédure pas à pas

Pour exécuter cet exemple, nous utilisons un notebook Python exécuté avec un noyau Data Science dans un environnement Studio. Vous pouvez également l'exécuter sur une instance de bloc-notes Python localement sur votre machine en configurant les informations d'identification pour assumer le rôle d'exécution SageMaker. Le notebook est léger et peut être exécuté sur une instance t3 medium par exemple. Des instructions détaillées étape par étape se trouvent dans le GitHub référentiel.

Vous pouvez soit utiliser la fonction d'exportation dans Data Wrangler pour générer le code Pipelines et le modifier pour Step Functions, soit créer votre propre script à partir de zéro. Dans notre exemple de référentiel, nous utilisons une combinaison des deux approches pour plus de simplicité. De manière générale, les étapes suivantes permettent de créer et d'exécuter le flux de travail Step Functions :

1. Générez un fichier de flux à partir de Data Wrangler ou utilisez le script de configuration pour générer un fichier de flux à partir d'un modèle préconfiguré.
2. Créez un compartiment S3 et chargez votre fichier de flux et vos fichiers d'entrée dans le compartiment.
3. Configurez votre rôle d'exécution SageMaker avec les autorisations requises comme mentionné précédemment. Reportez-vous au référentiel GitHub pour obtenir des instructions détaillées.
4. Suivez les instructions pour exécuter le bloc-notes dans le référentiel afin de démarrer un workflow. La tâche de traitement s'exécute sur un environnement Spark géré par SageMaker et peut prendre quelques minutes.
5. Accédez à la console Step Functions et suivez visuellement le flux de travail. Vous pouvez également accéder aux journaux CloudWatch liés pour déboguer les erreurs.

Passons en revue quelques sections importantes du code ici. Pour définir le flux de travail, nous définissons d'abord les étapes du flux de travail pour la machine d'état Step Function. La première étape est la data_wrangler_step pour le traitement des données, qui utilise le fichier de flux Data Wrangler comme entrée pour transformer les fichiers de données brutes. Nous définissons également une étape d'apprentissage du modèle et une étape de création du modèle nommée training_step ainsi que model_step, respectivement. Enfin, nous créons un workflow en enchaînant toutes les étapes que nous avons créées, comme illustré dans le code suivant :

from stepfunctions.steps import Chain 
from stepfunctions.workflow import Workflow 
import uuid 

workflow_graph = Chain([data_wrangler_step, training_step,model_step ]) 
branching_workflow = Workflow( name = "Wrangler-SF-Run-{}".format(uuid.uuid1().hex),definition = workflow_graph, role = iam_role ) 
branching_workflow.create()

Dans notre exemple, nous avons construit le flux de travail pour prendre les noms de travail comme paramètres, car ils sont uniques et doivent être générés de manière aléatoire lors de chaque exécution du pipeline. Nous transmettons ces noms lors de l'exécution du workflow. Vous pouvez également planifier l'exécution du workflow Step Functions à l'aide de CloudWatch (voir Planifier un flux de travail sans serveur avec AWS Step Functions et Amazon CloudWatch), invoqué en utilisant Événements Amazon S3, ou invoqué de Amazon Event Bridge (voir Créer une règle EventBridge qui déclenche un workflow Step Functions). À des fins de démonstration, nous pouvons appeler le flux de travail Step Functions à partir de l'interface utilisateur de la console Step Functions ou en utilisant le code suivant à partir du bloc-notes.

# Execute workflow 
execution = branching_workflow.execute( 
			inputs=  { “ProcessingJobName”: processing_job_name, # Unique processing job name, 
				   “TrainingJobName”: training_job_name, # Unique training job name, 
				   “ModelName” : model_name # Unique model name 
				 } 
	) 
execution_output = execution.get_output(wait=True)

Intégrer Data Wrangler à Apache Airflow

Un autre moyen populaire de créer des flux de travail ML consiste à utiliser Apache Airflow. Apache Airflow est une plate-forme open source qui vous permet de créer, de planifier et de surveiller par programmation des flux de travail. Amazon MWAA facilite la configuration et l'exploitation de pipelines ML de bout en bout avec Apache Airflow dans le cloud à grande échelle. Un pipeline Airflow consiste en une séquence de tâches, également appelées workflow. Un flux de travail est défini comme un DAG qui encapsule les tâches et les dépendances entre elles, définissant comment elles doivent s'exécuter dans le flux de travail.

Nous avons créé un DAG Airflow dans un environnement Amazon MWAA pour mettre en œuvre notre workflow MLOps. Chaque tâche du flux de travail est une unité exécutable, écrite en langage de programmation Python, qui exécute une action. Une tâche peut être soit une opérateur ou capteur. Dans notre cas, nous utilisons un Opérateur Python Airflow avec Kit de développement logiciel (SDK) SageMaker Python pour exécuter le script Data Wrangler Python et utiliser la prise en charge native d'Airflow Opérateurs SageMaker pour former l'algorithme XGBoost intégré de SageMaker et créer le modèle à partir des artefacts résultants. Nous avons également créé un opérateur Data Wrangler personnalisé utile (SageMakerDataWranglerOperator) pour Apache Airflow, que vous pouvez utiliser pour traiter les fichiers de flux Data Wrangler pour le traitement des données sans avoir besoin de code supplémentaire.

La capture d'écran suivante montre le DAG Airflow avec cinq étapes pour mettre en œuvre notre flux de travail MLOps.

L'étape Démarrer utilise un opérateur Python pour initialiser les configurations pour le reste des étapes du workflow. SageMaker_DataWrangler_Step Usages SageMakerDataWranglerOperator et le fichier de flux de données que nous avons créé précédemment. SageMaker_training_step ainsi que SageMaker_create_model_step utilisez les opérateurs SageMaker intégrés pour la formation et la création de modèles, respectivement. Notre environnement Amazon MWAA utilise le plus petit type d'instance (mw1.small), car la majeure partie du traitement est effectuée via des tâches de traitement, qui utilise son propre type d'instance qui peut être défini en tant que paramètres de configuration dans le flux de travail.

Installation et procédure pas à pas

Des instructions d'installation détaillées étape par étape pour déployer cette solution sont disponibles dans notre GitHub référentiel. Nous avons utilisé un notebook Jupyter avec des cellules de code Python pour configurer le DAG Airflow. En supposant que vous avez déjà généré le fichier de flux de données, voici une vue d'ensemble de haut niveau des étapes d'installation :

1. Créer un compartiment S3 et les dossiers suivants requis par Amazon MWAA.
2. Créer un environnement Amazon MWAA. Notez que nous avons utilisé Airflow version 2.0.2 pour cette solution.
3. Créer et télécharger un requirements.txt fichier avec toutes les dépendances Python requises par les tâches Airflow et chargez-le sur le /requirements répertoire dans le compartiment S3 principal Amazon MWAA. Il est utilisé par l'environnement Airflow géré pour installer les dépendances Python.
4. Télécharger le SMDataWranglerOperator.py fichier au /dags annuaire. Ce script Python contient du code pour l'opérateur Airflow personnalisé pour Data Wrangler. Cet opérateur peut être utilisé pour les tâches de traitement de n'importe quel fichier .flow.
5. Créer et télécharger le config.py scénario au /dags annuaire. Ce script Python est utilisé pour la première étape de notre DAG pour créer les objets de configuration requis par les étapes restantes du workflow.
6. Enfin, créez et téléchargez le ml_pipelines.py fichier au /dags annuaire. Ce script contient la définition du DAG pour le workflow Airflow. C'est là que nous définissons chacune des tâches et établissons des dépendances entre elles. Amazon MWAA interroge périodiquement le /dags pour exécuter ce script afin de créer le DAG ou de mettre à jour celui existant avec les dernières modifications.

Voici le code pour SageMaker_DataWrangler_step, qui utilise la coutume SageMakerDataWranglerOperator. Avec seulement quelques lignes de code dans votre script Python de définition DAG, vous pouvez pointer le SageMakerDataWranglerOperator à l'emplacement du fichier de flux Data Wrangler (qui est un emplacement S3). Dans les coulisses, cet opérateur utilise Emplois de traitement SageMaker pour traiter le fichier .flow afin d'appliquer les transformations définies à vos fichiers de données brutes. Vous pouvez également spécifier le type d'instance et le nombre d'instances nécessaires à la tâche de traitement Data Wrangler.

# Airflow Data Wrangler operator 
from SMDataWranglerOperator import SageMakerDataWranglerOperator 
preprocess_task = SageMakerDataWranglerOperator( task_id='DataWrangler_Processing_Step', 
                                                 dag=dag, 
                                                 flow_file_s3uri = flow_uri, 
                                                 processing_instance_count=2, 
                                                 instance_type='ml.m5.4xlarge', 
                                                 aws_conn_id="aws_default", 
                                                 config=config)

La config le paramètre accepte un dictionnaire (paires clé-valeur) de configurations supplémentaires requises par la tâche de traitement, telles que le préfixe de sortie du fichier de sortie final, le type de fichier de sortie (CSV ou Parquet) et l'URI pour le conteneur Data Wrangler intégré image. Le code suivant est ce que le config dictionnaire pour SageMakerDataWranglerOperator ressemble à. Ces configurations sont requises pour un traitement SageMaker processeur. Pour plus de détails sur chacun de ces paramètres de configuration, reportez-vous à sagemaker.processing.Processeur().

{
	"sagemaker_role": #required SageMaker IAM Role name or ARN,
	"s3_data_type": #optional;defaults to "S3Prefix"
	"s3_input_mode": #optional;defaults to "File",
	"s3_data_distribution_type": #optional;defaults to "FullyReplicated",
	"kms_key": #optional;defaults to None,
	"volume_size_in_gb": #optional;defaults to 30,
	"enable_network_isolation": #optional;defaults to False,
	"wait_for_processing": #optional;defaults to True,
	"container_uri": #optional;defaults to built - in container URI,
	"container_uri_pinned": #optional;defaults to built - in container URI,
	"outputConfig": {
		"s3_output_upload_mode": #optional;defaults to EndOfJob
		"output_content_type": #optional;defaults to CSV
		"output_bucket": #optional;defaults to SageMaker Default bucket
		"output_prefix": #optional;defaults to None.Prefix within bucket where output will be written
	}
}

Nettoyer

Pour éviter des frais futurs, supprimez les ressources créées pour les solutions que vous avez implémentées.

1. Suivez ces Des instructions fourni dans le référentiel GitHub pour nettoyer les ressources créées par la solution SageMaker Pipelines.
2. Suivez ces Des instructions fourni dans le référentiel GitHub pour nettoyer les ressources créées par la solution Step Functions.
3. Suivez ces Des instructions fourni dans le référentiel GitHub pour nettoyer les ressources créées par la solution Amazon MWAA.

Conclusion

Cet article a démontré comment vous pouvez facilement intégrer Data Wrangler à certaines des technologies d'automatisation et d'orchestration de flux de travail bien connues dans AWS. Nous avons d'abord examiné un exemple de cas d'utilisation et d'architecture pour la solution qui utilise Data Wrangler pour le prétraitement des données. Nous avons ensuite montré comment vous pouvez intégrer Data Wrangler avec Pipelines, Step Functions et Amazon MWAA.

Dans une prochaine étape, vous pouvez trouver et essayer les exemples de code et les blocs-notes dans notre GitHub référentiel en utilisant les instructions détaillées pour chacune des solutions décrites dans cet article. Pour en savoir plus sur la façon dont Data Wrangler peut aider vos charges de travail ML, visitez le Data Wrangler page produit ainsi que Préparer les données de ML avec Amazon SageMaker Data Wrangler.

À propos des auteurs

Rodrigo Alarcón est un architecte senior de solutions de stratégie ML avec AWS basé à Santiago, au Chili. Dans son rôle, Rodrigo aide les entreprises de différentes tailles à générer des résultats commerciaux grâce à des solutions d'IA et de ML basées sur le cloud. Ses intérêts incluent l'apprentissage automatique et la cybersécurité.

Ganapathi Krishnamoorthi est architecte senior de solutions ML chez AWS. Ganapathi fournit des conseils normatifs aux startups et aux entreprises clientes en les aidant à concevoir et à déployer des applications cloud à grande échelle. Il est spécialisé dans l'apprentissage automatique et s'attache à aider les clients à tirer parti de l'IA/ML pour leurs résultats commerciaux. Lorsqu'il n'est pas au travail, il aime explorer le plein air et écouter de la musique.

Anjan Biswas est un architecte senior de solutions de services d'IA spécialisé dans l'IA/ML et l'analyse de données. Anjan fait partie de l'équipe mondiale des services d'IA et travaille avec les clients pour les aider à comprendre et à développer des solutions aux problèmes commerciaux liés à l'IA et au ML. Anjan a plus de 14 ans d'expérience de travail avec des organisations mondiales de chaîne d'approvisionnement, de fabrication et de vente au détail et aide activement les clients à démarrer et à évoluer sur les services AWS AI.