Integrer Amazon SageMaker Data Wrangler med MLOps arbejdsgange

Efterhånden som virksomheder bevæger sig fra at køre ad hoc maskinlæringsmodeller (ML) til at bruge AI/ML til at transformere deres virksomhed i skala, bliver adoptionen af ​​ML Operations (MLOps) uundgåelig. Som vist i den følgende figur begynder ML-livscyklussen med at indramme et forretningsproblem som en ML-brugscase efterfulgt af en række faser, herunder dataforberedelse, funktionsudvikling, modelbygning, implementering, kontinuerlig overvågning og genoptræning. For mange virksomheder er mange af disse trin stadig manuelle og løst integreret med hinanden. Derfor er det vigtigt at automatisere end-to-end ML-livscyklussen, som muliggør hyppige eksperimenter for at skabe bedre forretningsresultater. Dataforberedelse er et af de afgørende trin i denne livscyklus, fordi ML-modellens nøjagtighed afhænger af kvaliteten af ​​træningsdatasættet.

Dataforskere og ML-ingeniører bruger et sted mellem 70-80 % af deres tid på at indsamle, analysere, rense og transformere data, der kræves til modeltræning. Amazon SageMaker Data Wrangler er en fuldt administreret kapacitet af Amazon SageMaker det gør det hurtigere for dataforskere og ML-ingeniører at analysere og forberede data til deres ML-projekter med lidt eller ingen kode. Når det kommer til operationalisering af en end-to-end ML-livscyklus, er dataforberedelse næsten altid det første trin i processen. Da der er mange måder at bygge en end-to-end ML-pipeline på, diskuterer vi i dette indlæg, hvordan du nemt kan integrere Data Wrangler med nogle af de velkendte workflow-automatiserings- og orkestreringsteknologier.

Løsningsoversigt

I dette indlæg demonstrerer vi, hvordan brugere kan integrere dataforberedelse ved hjælp af Data Wrangler med Amazon SageMaker Pipelines, AWS-trinfunktionerog Apache luftstrøm med Amazon Managed Workflow for Apache Airflow (Amazon MWAA). Pipelines er en SageMaker-funktion, der er en specialbygget og brugervenlig kontinuerlig integration og kontinuerlig levering (CI/CD) service til ML. Step Functions er en serverløs, lav-kode visuel workflow-tjeneste, der bruges til at orkestrere AWS-tjenester og automatisere forretningsprocesser. Amazon MWAA er en administreret orkestreringstjeneste til Apache Airflow, der gør det nemmere at betjene end-to-end-data og ML-pipelines.

Til demonstrationsformål overvejer vi en use case til at forberede data til at træne en ML-model med SageMaker indbygget XGBoost algoritme som vil hjælpe os med at identificere svigagtige køretøjsforsikringskrav. Vi brugte et syntetisk genereret sæt eksempeldata til at træne modellen og oprette en SageMaker-model ved hjælp af modelartefakter fra træningsprocessen. Vores mål er at operationalisere denne proces fra ende til anden ved at opsætte et ML-workflow. Selvom ML-arbejdsgange kan være mere omfattende, bruger vi en minimal arbejdsgang til demonstrationsformål. Det første trin i arbejdsgangen er dataforberedelse med Data Wrangler, efterfulgt af et modeltræningstrin og til sidst et modeloprettelsestrin. Følgende diagram illustrerer vores løsningsarbejdsgang.

I de følgende afsnit gennemgår vi, hvordan du opsætter et Data Wrangler-flow og integrerer Data Wrangler med Pipelines, Step Functions og Apache Airflow.

Opsæt et Data Wrangler-flow

Vi starter med at lave et Data Wrangler flow, også kaldet en dataflowVed hjælp af dataflow UI via Amazon SageMaker Studio IDE. Vores eksempeldatasæt består af to datafiler: krav.csv , kunder.csv, som opbevares i en Amazon Simple Storage Service (Amazon S3) spand. Vi bruger dataflow UI til at anvende Data Wranglers indbyggede transformationer sådan kategorisk kodning, strengformatering og imputation til funktionskolonnerne i hver af disse filer. Vi ansøger også tilpasset transformation til et par funktionskolonner ved hjælp af et par linjer tilpasset Python-kode med Pandas DataFrame. Følgende skærmbillede viser transformationerne anvendt på claims.csv-filen i dataflow-brugergrænsefladen.

Til sidst samler vi resultaterne af de anvendte transformationer af de to datafiler for at generere et enkelt træningsdatasæt til vores modeltræning. Vi bruger Data Wranglers indbyggede tilslutte datasæt kapacitet, som lader os udføre SQL-lignende join-operationer på tabeldata. Følgende skærmbillede viser dataflowet i dataflow-brugergrænsefladen i Studio. For trin-for-trin instruktioner til at oprette dataflowet ved hjælp af Data Wrangler, se GitHub repository.

Du kan nu bruge dataflow-filen (.flow) til at udføre datatransformationer på vores rådatafiler. Dataflow-brugergrænsefladen kan automatisk generere Python-notebooks, som vi kan bruge og integrere direkte med Pipelines ved hjælp af SageMaker SDK. Til trinfunktioner bruger vi AWS Step Funktioner Data Science Python SDK at integrere vores Data Wrangler-behandling med en Step Functions-pipeline. Til Amazon MWAA bruger vi en tilpasset luftstrømsoperatør og Airflow SageMaker operatør. Vi diskuterer hver af disse tilgange i detaljer i de følgende afsnit.

Integrer Data Wrangler med Pipelines

SageMaker Pipelines er et indbygget workflow-orkestreringsværktøj til at bygge ML-pipelines, der drager fordel af direkte SageMaker-integration. Sammen med SageMaker modelregistrering, Pipelines forbedrer den operationelle modstandsdygtighed og reproducerbarhed af dine ML-arbejdsgange. Disse workflow-automatiseringskomponenter giver dig mulighed for nemt at skalere din evne til at bygge, træne, teste og implementere hundredvis af modeller i produktionen; iterere hurtigere; reducere fejl på grund af manuel orkestrering; og opbygge gentagelige mekanismer. Hvert trin i pipelinen kan holde styr på afstamningen, og mellemtrin kan cachelagres for hurtigt at køre pipelinen igen. Du kan oprette pipelines ved hjælp af SageMaker Python SDK.

En arbejdsgang bygget med SageMaker-pipelines består af en sekvens af trin, der danner en Directed Acyclic Graph (DAG). I dette eksempel begynder vi med a forarbejdningstrin, som kører en SageMaker Processing job baseret på Data Wranglers flowfil for at oprette et træningsdatasæt. Så fortsætter vi med en træningstrin, hvor vi træner en XGBoost-model ved hjælp af SageMakers indbyggede XGBoost-algoritme og træningsdatasættet oprettet i forrige trin. Efter at en model er blevet trænet, afslutter vi denne arbejdsgang med en RegisterModel trin at registrere den trænede model med SageMaker-modelregistret.

Installation og gennemgang

For at køre denne prøve bruger vi en Jupyter-notesbog, der kører Python3 på et Data Science-kernebillede i et Studio-miljø. Du kan også køre det på en Jupyter notebook-instans lokalt på din maskine ved at konfigurere legitimationsoplysningerne til at påtage sig SageMaker-udførelsesrollen. Notebook'en er let og kan køre på en ml.t3.medium instans. Detaljerede trin-for-trin instruktioner kan findes i GitHub repository.

Du kan enten bruge eksportfunktionen i Data Wrangler til at generere Pipelines-koden eller bygge dit eget script fra bunden. I vores prøvelager bruger vi en kombination af begge tilgange for enkelhedens skyld. På et højt niveau er følgende trin til at opbygge og køre Pipelines-arbejdsgangen:

1. Generer en flowfil fra Data Wrangler, eller brug opsætningsscriptet til at generere en flowfil fra en forudkonfigureret skabelon.
2. Opret en Amazon Simple Storage Service (Amazon S3) bucket og upload din flow-fil og input-filer til bucket. I vores eksempelnotesbog bruger vi SageMaker standard S3-bøtten.
3. Følg instruktionerne i notesbogen for at oprette et Processor-objekt baseret på Data Wrangler-flowfilen og et Estimator-objekt med parametrene for træningsjobbet.
   1. I vores eksempel, fordi vi kun bruger SageMaker-funktioner og standard S3-bøtten, kan vi bruge Studios standardudførelsesrolle. Det samme AWS identitets- og adgangsstyring (IAM)-rollen påtages af pipeline-kørslen, behandlingsjobbet og træningsjobbet. Du kan yderligere tilpasse udførelsesrollen i henhold til minimumsrettigheder.
4. Fortsæt med instruktionerne for at oprette en pipeline med trin, der refererer til Processor- og Estimator-objekterne, og kør derefter pipelinen. Behandlings- og træningsopgaverne kører på SageMaker-administrerede miljøer og tager et par minutter at fuldføre.
5. I Studio kan du se pipelinedetaljerne overvåge pipelinekørslen. Du kan også overvåge de underliggende behandlings- og træningsjob fra SageMaker-konsollen eller fra amazoncloudwatch.

Integrer Data Wrangler med Step-funktioner

Med Step Functions kan du udtrykke kompleks forretningslogik som lavkodede, hændelsesdrevne arbejdsgange, der forbinder forskellige AWS-tjenester. Det Trinfunktioner Data Science SDK er et open source-bibliotek, der giver datavidenskabsfolk mulighed for at skabe arbejdsgange, der kan forbehandle datasæt og bygge, implementere og overvåge ML-modeller ved hjælp af SageMaker og Step Functions. Trinfunktioner er baseret på tilstandsmaskiner og opgaver. Trinfunktioner opretter arbejdsgange ud af trin kaldet stater, og udtrykker den arbejdsgang i Amazon-staternes sprog. Når du opretter en arbejdsgang ved hjælp af Step Functions Data Science SDK, opretter den en tilstandsmaskine, der repræsenterer dit workflow og trin i Step Functions.

Til denne brugssag har vi bygget en Step Functions-arbejdsgang baseret på det almindelige mønster, der bruges i dette indlæg, der inkluderer et behandlingstrin, træningstrin og RegisterModel trin. I dette tilfælde importerer vi disse trin fra Step Functions Data Science Python SDK. Vi sammenkæder disse trin i samme rækkefølge for at skabe et Step Functions-arbejdsforløb. Workflowet bruger flowfilen, der blev genereret fra Data Wrangler, men du kan også bruge din egen Data Wrangler flowfil. Vi genbruger noget kode fra Data Wrangler eksportfunktion for enkelhedens skyld. Vi kører dataforbehandlingslogikken genereret af Data Wrangler-flowfilen for at skabe et træningsdatasæt, træne en model ved hjælp af XGBoost-algoritmen og gemme den trænede modelartefakt som en SageMaker-model. Derudover viser vi i GitHub-repoen også, hvordan Step Functions giver os mulighed for at prøve at fange fejl og håndtere fejl og genforsøg med FailStateStep , CatchStateStep.

Det resulterende flowdiagram, som vist i det følgende skærmbillede, er tilgængeligt på konsollen Trinfunktioner, efter at arbejdsgangen er startet. Dette hjælper datavidenskabsfolk og ingeniører med at visualisere hele workflowet og hvert trin i det og få adgang til de linkede CloudWatch-logfiler for hvert trin.

Installation og gennemgang

For at køre denne prøve bruger vi en Python-notesbog, der kører med en Data Science-kerne i et Studio-miljø. Du kan også køre det på en Python-notebook-instans lokalt på din maskine ved at konfigurere legitimationsoplysningerne til at påtage sig SageMaker-udførelsesrollen. Notebook'en er let og kan f.eks. køres på en t3 medium. Detaljerede trin-for-trin instruktioner kan findes i GitHub repository.

Du kan enten bruge eksportfunktionen i Data Wrangler til at generere Pipelines-koden og ændre den til Step Functions eller bygge dit eget script fra bunden. I vores prøvelager bruger vi en kombination af begge tilgange for enkelhedens skyld. På et højt niveau er følgende trin til at opbygge og køre Step Functions-arbejdsgangen:

1. Generer en flowfil fra Data Wrangler, eller brug opsætningsscriptet til at generere en flowfil fra en forudkonfigureret skabelon.
2. Opret en S3-bøtte og upload din flowfil og inputfiler til bøtten.
3. Konfigurer din SageMaker-udførelsesrolle med de nødvendige tilladelser som nævnt tidligere. Se GitHub-lageret for detaljerede instruktioner.
4. Følg instruktionerne for at køre notesbogen i lageret for at starte en arbejdsgang. Behandlingsjobbet kører på et SageMaker-administreret Spark-miljø og kan tage få minutter at fuldføre.
5. Gå til Step Functions-konsollen og spor workflowet visuelt. Du kan også navigere til de tilknyttede CloudWatch-logfiler for at fejlfinde.

Lad os gennemgå nogle vigtige dele af koden her. For at definere workflowet definerer vi først trinene i workflowet for Step Function-tilstandsmaskinen. Det første skridt er data_wrangler_step til databehandling, som bruger Data Wrangler-flowfilen som input til at transformere de rå datafiler. Vi definerer også et modeltræningstrin og et modeloprettelsestrin med navn training_step , model_step, henholdsvis. Til sidst opretter vi en arbejdsgang ved at sammenkæde alle de trin, vi har oprettet, som vist i følgende kode:

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()

I vores eksempel byggede vi arbejdsgangen til at tage jobnavne som parametre, fordi de er unikke og skal genereres tilfældigt under hver pipelinekørsel. Vi videregiver disse navne, når arbejdsgangen kører. Du kan også planlægge Step Functions workflow til at køre ved hjælp af CloudWatch (se Planlæg et serverløst workflow med AWS Step Functions og Amazon CloudWatch), påberåbt vha Amazon S3-begivenheder, eller påberåbes fra Amazon Eventbridge (Se Opret en EventBridge-regel, der udløser et Step Functions-arbejdsforløb). Til demonstrationsformål kan vi kalde Step Functions-arbejdsgangen fra Step Functions-konsollens brugergrænseflade eller ved at bruge følgende kode fra notesbogen.

# 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)

Integrer Data Wrangler med Apache Airflow

En anden populær måde at skabe ML-arbejdsgange på er at bruge Apache Airflow. Apache Airflow er en open source-platform, der giver dig mulighed for programmæssigt at oprette, planlægge og overvåge arbejdsgange. Amazon MWAA gør det nemt at opsætte og betjene end-to-end ML-pipelines med Apache Airflow i skyen i stor skala. En luftstrømsrørledning består af en række opgaver, også kaldet en workflow. En arbejdsgang er defineret som en DAG, der indkapsler opgaverne og afhængighederne mellem dem og definerer, hvordan de skal køre i arbejdsgangen.

Vi oprettede en Airflow DAG i et Amazon MWAA-miljø for at implementere vores MLOps-arbejdsgang. Hver opgave i arbejdsgangen er en eksekverbar enhed, skrevet i Python-programmeringssproget, der udfører en eller anden handling. En opgave kan enten være en operatør eller sensor. I vores tilfælde bruger vi en Airflow Python-operatør sammen med SageMaker Python SDK at køre Data Wrangler Python-scriptet og bruge Airflow's oprindeligt understøttede SageMaker operatører at træne den SageMaker indbyggede XGBoost-algoritme og skabe modellen ud fra de resulterende artefakter. Vi har også oprettet en nyttig brugerdefineret Data Wrangler-operatør (SageMakerDataWranglerOperator) til Apache Airflow, som du kan bruge til at behandle Data Wrangler-flowfiler til databehandling uden behov for yderligere kode.

Følgende skærmbillede viser Airflow DAG med fem trin til implementering af vores MLOps-arbejdsgang.

Start-trinnet bruger en Python-operator til at initialisere konfigurationer for resten af ​​trinene i arbejdsgangen. SageMaker_DataWrangler_Step bruger SageMakerDataWranglerOperator og dataflowfilen, vi oprettede tidligere. SageMaker_training_step , SageMaker_create_model_step bruge de indbyggede SageMaker-operatører til henholdsvis modeltræning og modeloprettelse. Vores Amazon MWAA-miljø bruger den mindste instanstype (mw1.small), fordi hovedparten af ​​behandlingen sker via Processing jobs, som bruger sin egen instanstype, der kan defineres som konfigurationsparametre i arbejdsgangen.

Installation og gennemgang

Detaljerede trin-for-trin installationsinstruktioner til implementering af denne løsning kan findes i vores GitHub repository. Vi brugte en Jupyter-notesbog med Python-kodeceller til at konfigurere Airflow DAG. Forudsat at du allerede har genereret dataflowfilen, er følgende en oversigt på højt niveau over installationstrinnene:

1. Opret en S3-spand og efterfølgende mapper, der kræves af Amazon MWAA.
2. Opret et Amazon MWAA-miljø. Bemærk, at vi brugte Airflow version 2.0.2 til denne løsning.
3. Opret og upload en requirements.txt fil med alle de Python-afhængigheder, der kræves af Airflow-opgaverne, og upload den til /requirements bibliotek i Amazons MWAA primære S3-spand. Dette bruges af det administrerede Airflow-miljø til at installere Python-afhængighederne.
4. Upload SMDataWranglerOperator.py fil til /dags vejviser. Dette Python-script indeholder kode til den tilpassede Airflow-operatør til Data Wrangler. Denne operatør kan bruges til opgaver til at behandle enhver .flow-fil.
5. Opret og upload config.py script til /dags vejviser. Dette Python-script bruges til det første trin i vores DAG til at skabe konfigurationsobjekter, der kræves af de resterende trin i arbejdsgangen.
6. Til sidst skal du oprette og uploade ml_pipelines.py fil til /dags vejviser. Dette script indeholder DAG-definitionen for Airflow-arbejdsgangen. Det er her, vi definerer hver af opgaverne og opsætter afhængigheder mellem dem. Amazon MWAA afstemninger med jævne mellemrum /dags mappe for at køre dette script for at oprette DAG eller opdatere den eksisterende med eventuelle seneste ændringer.

Følgende er koden til SageMaker_DataWrangler_step, som bruger skik og brug SageMakerDataWranglerOperator. Med blot et par linjer kode i dit DAG definition Python script, kan du pege på SageMakerDataWranglerOperator til Data Wrangler-flowfilplaceringen (som er en S3-placering). Bag kulisserne bruger denne operatør SageMaker Processing jobs at behandle .flow-filen for at anvende de definerede transformationer til dine rådatafiler. Du kan også angive type instans og antallet af forekomster, der kræves af Data Wrangler-behandlingsjobbet.

# 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)

config parameter accepterer en ordbog (nøgleværdi-par) af yderligere konfigurationer, der kræves af behandlingsjobbet, såsom outputpræfikset for den endelige outputfil, type outputfil (CSV eller Parquet) og URI for den indbyggede Data Wrangler-beholder billede. Følgende kode er hvad config ordbog til SageMakerDataWranglerOperator ligner. Disse konfigurationer er nødvendige for en SageMaker-behandling processor. For detaljer om hver af disse konfigurationsparametre, se sagemaker.processing.Processor().

{
	"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
	}
}

Ryd op

For at undgå fremtidige gebyrer skal du slette de ressourcer, der er oprettet for de løsninger, du implementerede.

1. Følg disse anvisninger leveret i GitHub-lageret for at rydde op i ressourcer skabt af SageMaker Pipelines-løsningen.
2. Følg disse anvisninger leveret i GitHub-lageret for at rydde op i ressourcer, der er oprettet af Step Functions-løsningen.
3. Følg disse anvisninger leveres i GitHub-lageret for at rydde op i ressourcer skabt af Amazon MWAA-løsningen.

Konklusion

Dette indlæg demonstrerede, hvordan du nemt kan integrere Data Wrangler med nogle af de velkendte workflow-automatiserings- og orkestreringsteknologier i AWS. Vi gennemgik først et eksempel på use case og arkitektur for løsningen, der bruger Data Wrangler til dataforbehandling. Derefter demonstrerede vi, hvordan du kan integrere Data Wrangler med Pipelines, Step Functions og Amazon MWAA.

Som et næste trin kan du finde og prøve kodeeksemplerne og notesbøgerne i vores GitHub repository ved hjælp af de detaljerede instruktioner for hver af de løsninger, der er diskuteret i dette indlæg. Besøg Data Wrangler for at lære mere om, hvordan Data Wrangler kan hjælpe dine ML-arbejdsbelastninger Produkt side , Forbered ML Data med Amazon SageMaker Data Wrangler.

Om forfatterne

Rodrigo Alarcon er en senior ML Strategy Solutions Architect med AWS baseret i Santiago, Chile. I sin rolle hjælper Rodrigo virksomheder af forskellige størrelser med at generere forretningsresultater gennem cloud-baserede AI- og ML-løsninger. Hans interesser omfatter maskinlæring og cybersikkerhed.

Ganapathi Krishnamoorthi er Senior ML Solutions Architect hos AWS. Ganapathi giver foreskrivende vejledning til startup- og virksomhedskunder, der hjælper dem med at designe og implementere cloud-applikationer i stor skala. Han er specialiseret i maskinlæring og er fokuseret på at hjælpe kunder med at udnytte AI/ML til deres forretningsresultater. Når han ikke er på arbejde, nyder han at udforske udendørs og lytte til musik.

Anjan Biswas er Senior AI Services Solutions Architect med fokus på AI/ML og Data Analytics. Anjan er en del af det verdensomspændende AI-serviceteam og arbejder med kunder for at hjælpe dem med at forstå og udvikle løsninger på forretningsproblemer med AI og ML. Anjan har over 14 års erfaring med at arbejde med globale forsyningskæder, fremstillings- og detailorganisationer og hjælper aktivt kunder med at komme i gang og skalere på AWS AI-tjenester.