Integrer Amazon SageMaker Data Wrangler med MLOps arbeidsflyter

Etter hvert som bedrifter går fra å kjøre ad hoc maskinlæringsmodeller (ML) til å bruke AI/ML for å transformere virksomheten i stor skala, blir bruken av ML Operations (MLOps) uunngåelig. Som vist i følgende figur, begynner ML-livssyklusen med å utforme et forretningsproblem som en ML-brukssak, etterfulgt av en rekke faser, inkludert dataforberedelse, funksjonsutvikling, modellbygging, distribusjon, kontinuerlig overvåking og omskolering. For mange bedrifter er mange av disse trinnene fortsatt manuelle og løst integrert med hverandre. Derfor er det viktig å automatisere ende-til-ende ML-livssyklusen, som muliggjør hyppige eksperimenter for å oppnå bedre forretningsresultater. Dataforberedelse er et av de avgjørende trinnene i denne livssyklusen, fordi ML-modellens nøyaktighet avhenger av kvaliteten på opplæringsdatasettet.

Dataforskere og ML-ingeniører bruker et sted mellom 70–80 % av tiden sin på å samle inn, analysere, rense og transformere data som kreves for modelltrening. Amazon SageMaker Data Wrangler er en fullstendig administrert evne til Amazon SageMaker som gjør det raskere for dataforskere og ML-ingeniører å analysere og forberede data for sine ML-prosjekter med lite eller ingen kode. Når det gjelder operasjonalisering av en ende-til-ende ML-livssyklus, er dataforberedelse nesten alltid det første trinnet i prosessen. Gitt at det er mange måter å bygge en ende-til-ende ML-pipeline på, diskuterer vi i dette innlegget hvordan du enkelt kan integrere Data Wrangler med noen av de velkjente arbeidsflytautomatiserings- og orkestreringsteknologiene.

Løsningsoversikt

I dette innlegget viser vi hvordan brukere kan integrere dataforberedelse ved å bruke Data Wrangler med Amazon SageMaker-rørledninger, AWS trinnfunksjonerog Apache luftstrøm med Amazon Managed Workflow for Apache Airflow (Amazon MWAA). Pipelines er en SageMaker-funksjon som er en spesialbygd og brukervennlig kontinuerlig integrasjon og kontinuerlig levering (CI/CD)-tjeneste for ML. Step Functions er en serverløs, lavkode visuell arbeidsflyttjeneste som brukes til å orkestrere AWS-tjenester og automatisere forretningsprosesser. Amazon MWAA er en administrert orkestreringstjeneste for Apache Airflow som gjør det enklere å betjene ende-til-ende-data og ML-rørledninger.

For demonstrasjonsformål vurderer vi en brukssak for å forberede data for å trene en ML-modell med SageMaker innebygd XGBoost-algoritme som vil hjelpe oss med å identifisere uredelige kjøretøyforsikringskrav. Vi brukte et syntetisk generert sett med eksempeldata for å trene modellen og lage en SageMaker-modell ved å bruke modellartefaktene fra treningsprosessen. Målet vårt er å operasjonalisere denne prosessen fra ende til annen ved å sette opp en ML-arbeidsflyt. Selv om ML-arbeidsflyt kan være mer forseggjort, bruker vi en minimal arbeidsflyt for demonstrasjonsformål. Det første trinnet i arbeidsflyten er dataforberedelse med Data Wrangler, etterfulgt av et modelltreningstrinn, og til slutt et trinn for modelloppretting. Følgende diagram illustrerer arbeidsflyten vår for løsningen.

I de følgende delene viser vi deg hvordan du setter opp en Data Wrangler-flyt og integrerer Data Wrangler med Pipelines, Step Functions og Apache Airflow.

Sett opp en Data Wrangler-flyt

Vi starter med å lage en Data Wrangler-flyt, også kalt en dataflyt, bruker dataflyt UI via Amazon SageMaker Studio IDE. Eksempeldatasettet vårt består av to datafiler: claims.csv og kunder.csv, som er lagret i en Amazon enkel lagringstjeneste (Amazon S3) bøtte. Vi bruker dataflyt-UI for å bruke Data Wranglers innebygde transformasjoner slik kategorisk koding, strengformatering og imputering til funksjonskolonnene i hver av disse filene. Vi søker også tilpasset transformasjon til noen få funksjonskolonner ved å bruke noen få linjer med egendefinert Python-kode med Pandas DataFrame. Følgende skjermbilde viser transformasjonene brukt på claims.csv-filen i dataflyt-UI.

Til slutt slår vi sammen resultatene av de anvendte transformasjonene av de to datafilene for å generere et enkelt opplæringsdatasett for vår modelltrening. Vi bruker Data Wranglers innebygde bli med i datasett funksjon, som lar oss utføre SQL-lignende sammenføyningsoperasjoner på tabelldata. Følgende skjermbilde viser dataflyten i dataflyt-UI i Studio. For trinnvise instruksjoner for å lage dataflyten ved hjelp av Data Wrangler, se GitHub repository.

Du kan nå bruke dataflytfilen (.flow) til å utføre datatransformasjoner på våre rådatafiler. Dataflyt-UI kan automatisk generere Python-notatbøker som vi kan bruke og integrere direkte med Pipelines ved å bruke SageMaker SDK. For trinnfunksjoner bruker vi AWS Step Functions Data Science Python SDK å integrere vår Data Wrangler-behandling med en Step Functions-pipeline. For Amazon MWAA bruker vi en tilpasset luftstrømsoperatør og Airflow SageMaker-operatør. Vi diskuterer hver av disse tilnærmingene i detalj i de følgende avsnittene.

Integrer Data Wrangler med Pipelines

SageMaker Pipelines er et naturlig orkestreringsverktøy for arbeidsflyt for å bygge ML-pipelines som drar fordel av direkte SageMaker-integrasjon. Sammen med SageMaker modellregister, Pipelines forbedrer den operasjonelle motstandskraften og reproduserbarheten til ML-arbeidsflytene dine. Disse automatiseringskomponentene for arbeidsflyt lar deg enkelt skalere din evne til å bygge, trene, teste og distribuere hundrevis av modeller i produksjon; iterere raskere; redusere feil på grunn av manuell orkestrering; og bygge repeterbare mekanismer. Hvert trinn i rørledningen kan holde styr på avstamningen, og mellomtrinn kan bufres for raskt å kjøre rørledningen på nytt. Du kan lage rørledninger ved hjelp av SageMaker Python SDK.

En arbeidsflyt bygget med SageMaker-rørledninger består av en sekvens av trinn som danner en Directed Acyclic Graph (DAG). I dette eksemplet begynner vi med a behandlingstrinn, som kjører en SageMaker Behandler jobb basert på Data Wranglers flytfil for å lage et treningsdatasett. Vi fortsetter så med en treningstrinn, hvor vi trener en XGBoost-modell ved hjelp av SageMakers innebygde XGBoost-algoritme og treningsdatasettet opprettet i forrige trinn. Etter at en modell er opplært, avslutter vi denne arbeidsflyten med en RegisterModel trinn for å registrere den trente modellen med SageMaker modellregister.

Installasjon og gjennomgang

For å kjøre denne prøven bruker vi en Jupyter-notisbok som kjører Python3 på et Data Science-kjernebilde i et Studio-miljø. Du kan også kjøre den på en Jupyter-notebook-forekomst lokalt på maskinen din ved å sette opp legitimasjonen for å påta seg SageMaker-utførelsesrollen. Den bærbare datamaskinen er lett og kan kjøres på en ml.t3.medium-instans. Detaljerte trinnvise instruksjoner finner du i GitHub repository.

Du kan enten bruke eksportfunksjonen i Data Wrangler for å generere Pipelines-koden, eller bygge ditt eget skript fra bunnen av. I vårt prøvelager bruker vi en kombinasjon av begge tilnærmingene for enkelhets skyld. På et høyt nivå er følgende trinn for å bygge og kjøre Pipelines-arbeidsflyten:

1. Generer en flytfil fra Data Wrangler eller bruk oppsettskriptet til å generere en flytfil fra en forhåndskonfigurert mal.
2. Lag en Amazon enkel lagringstjeneste (Amazon S3) bøtte og last opp flytfilen og inndatafilene til bøtten. I vår eksempelnotisbok bruker vi SageMaker standard S3-bøtte.
3. Følg instruksjonene i notatboken for å lage et prosessorobjekt basert på Data Wrangler-flytfilen, og et estimatorobjekt med parameterne for treningsjobben.
   1. I vårt eksempel, fordi vi bare bruker SageMaker-funksjoner og standard S3-bøtte, kan vi bruke Studios standard utførelsesrolle. Det samme AWS identitets- og tilgangsadministrasjon (IAM)-rollen påtas av rørledningen, prosesseringsjobben og opplæringsjobben. Du kan tilpasse utførelsesrollen ytterligere i henhold til minimumsrettigheter.
4. Fortsett med instruksjonene for å lage en pipeline med trinn som viser til prosessor- og estimatorobjektene, og kjør deretter pipelinen. Behandlings- og opplæringsjobbene kjører på SageMaker administrerte miljøer og tar noen minutter å fullføre.
5. I Studio kan du se pipeline-detaljene overvåke pipeline-kjøringen. Du kan også overvåke de underliggende prosesserings- og opplæringsjobbene fra SageMaker-konsollen eller fra Amazon CloudWatch.

Integrer Data Wrangler med trinnfunksjoner

Med Step Functions kan du uttrykke kompleks forretningslogikk som lavkode, hendelsesdrevne arbeidsflyter som kobler sammen ulike AWS-tjenester. De Trinnfunksjoner Data Science SDK er et åpen kildekode-bibliotek som lar dataforskere lage arbeidsflyter som kan forhåndsbehandle datasett og bygge, distribuere og overvåke ML-modeller ved hjelp av SageMaker og Step Functions. Step Functions er basert på tilstandsmaskiner og oppgaver. Step Functions oppretter arbeidsflyter ut fra trinn kalt stater, og uttrykker den arbeidsflyten i Amazon State Language. Når du oppretter en arbeidsflyt ved hjelp av Step Functions Data Science SDK, oppretter den en tilstandsmaskin som representerer arbeidsflyten og trinnene i Step Functions.

For denne brukssaken har vi bygget en Step Functions-arbeidsflyt basert på det vanlige mønsteret som brukes i dette innlegget, som inkluderer et prosesseringstrinn, opplæringstrinn og RegisterModel steg. I dette tilfellet importerer vi disse trinnene fra Step Functions Data Science Python SDK. Vi lenker disse trinnene i samme rekkefølge for å lage en Step Functions-arbeidsflyt. Arbeidsflyten bruker flytfilen som ble generert fra Data Wrangler, men du kan også bruke din egen Data Wrangler flytfil. Vi gjenbruker noe kode fra Data Wrangler eksportfunksjon for enkelhets skyld. Vi kjører dataforbehandlingslogikken generert av Data Wrangler-flytfilen for å lage et treningsdatasett, trene en modell ved hjelp av XGBoost-algoritmen og lagre den trente modellartefakten som en SageMaker-modell. I tillegg viser vi i GitHub-repoen hvordan Step Functions lar oss prøve å fange opp feil, og håndtere feil og gjenforsøk med FailStateStep og CatchStateStep.

Det resulterende flytdiagrammet, som vist i følgende skjermbilde, er tilgjengelig på Step Functions-konsollen etter at arbeidsflyten har startet. Dette hjelper dataforskere og ingeniører med å visualisere hele arbeidsflyten og hvert trinn i den, og få tilgang til de koblede CloudWatch-loggene for hvert trinn.

Installasjon og gjennomgang

For å kjøre denne prøven bruker vi en Python-notatbok som kjører med en Data Science-kjerne i et Studio-miljø. Du kan også kjøre den på en Python-notebook-forekomst lokalt på maskinen din ved å sette opp legitimasjonen for å påta seg SageMaker-utførelsesrollen. Den bærbare datamaskinen er lett og kan for eksempel kjøres på en t3 medium. Detaljerte trinnvise instruksjoner finner du i GitHub repository.

Du kan enten bruke eksportfunksjonen i Data Wrangler til å generere Pipelines-koden og endre den for trinnfunksjoner eller bygge ditt eget skript fra bunnen av. I vårt prøvelager bruker vi en kombinasjon av begge tilnærmingene for enkelhets skyld. På et høyt nivå er følgende trinn for å bygge og kjøre arbeidsflyten for trinnfunksjoner:

1. Generer en flytfil fra Data Wrangler eller bruk oppsettskriptet til å generere en flytfil fra en forhåndskonfigurert mal.
2. Lag en S3-bøtte og last opp flytfilen og inndatafilene til bøtten.
3. Konfigurer SageMaker-utførelsesrollen din med de nødvendige tillatelsene som nevnt tidligere. Se GitHub-depotet for detaljerte instruksjoner.
4. Følg instruksjonene for å kjøre notatblokken i depotet for å starte en arbeidsflyt. Behandlingsjobben kjører på et SageMaker-administrert Spark-miljø og kan ta noen minutter å fullføre.
5. Gå til Step Functions-konsollen og spor arbeidsflyten visuelt. Du kan også navigere til de koblede CloudWatch-loggene for å feilsøke feil.

La oss se gjennom noen viktige deler av koden her. For å definere arbeidsflyten, definerer vi først trinnene i arbeidsflyten for Step Function-tilstandsmaskinen. Det første trinnet er data_wrangler_step for databehandling, som bruker Data Wrangler-flytfilen som input for å transformere rådatafilene. Vi definerer også et modelltreningstrinn og et modellopprettingstrinn kalt training_step og model_step, henholdsvis. Til slutt lager vi en arbeidsflyt ved å lenke alle trinnene vi opprettet, 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 vårt eksempel bygde vi arbeidsflyten for å ta jobbnavn som parametere fordi de er unike og må genereres tilfeldig under hver pipelinekjøring. Vi sender disse navnene når arbeidsflyten kjører. Du kan også planlegge Step Functions-arbeidsflyten til å kjøre ved hjelp av CloudWatch (se Planlegg en serverløs arbeidsflyt med AWS Step Functions og Amazon CloudWatch), påkalt ved hjelp av Amazon S3-arrangementer, eller påberopt fra Amazon EventBridge (Se Opprett en EventBridge-regel som utløser en Step Functions-arbeidsflyt). For demonstrasjonsformål kan vi starte Step Functions-arbeidsflyten fra Step Functions-konsollens brukergrensesnitt eller ved å bruke følgende kode fra notatboken.

# 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 annen populær måte å lage ML-arbeidsflyter på er å bruke Apache Airflow. Apache Airflow er en åpen kildekode-plattform som lar deg programmere forfattere, planlegge og overvåke arbeidsflyter. Amazon MWAA gjør det enkelt å sette opp og drifte ende-til-ende ML-rørledninger med Apache Airflow i skyen i stor skala. En luftstrømrørledning består av en sekvens av oppgaver, også referert til som en arbeidsflyt. En arbeidsflyt er definert som en DAG som innkapsler oppgavene og avhengighetene mellom dem, og definerer hvordan de skal kjøres i arbeidsflyten.

Vi opprettet en Airflow DAG i et Amazon MWAA-miljø for å implementere MLOps-arbeidsflyten vår. Hver oppgave i arbeidsflyten er en kjørbar enhet, skrevet i programmeringsspråket Python, som utfører en handling. En oppgave kan enten være en operatør eller en sensor. I vårt tilfelle bruker vi en Airflow Python-operatør samt SageMaker Python SDK for å kjøre Data Wrangler Python-skriptet, og bruke Airflow som støttes SageMaker-operatører å trene SageMaker innebygde XGBoost-algoritme og lage modellen fra de resulterende artefaktene. Vi har også laget en nyttig tilpasset Data Wrangler-operatør (SageMakerDataWranglerOperator) for Apache Airflow, som du kan bruke til å behandle data Wrangler-flytfiler for databehandling uten behov for tilleggskode.

Følgende skjermbilde viser Airflow DAG med fem trinn for å implementere MLOps-arbeidsflyten vår.

Start-trinnet bruker en Python-operatør for å initialisere konfigurasjoner for resten av trinnene i arbeidsflyten. SageMaker_DataWrangler_Step bruker SageMakerDataWranglerOperator og dataflytfilen vi opprettet tidligere. SageMaker_training_step og SageMaker_create_model_step bruk de innebygde SageMaker-operatørene for henholdsvis modellopplæring og modellskaping. Vårt Amazon MWAA-miljø bruker den minste forekomsttypen (mw1.small), fordi hoveddelen av behandlingen gjøres via Processing-jobber, som bruker sin egen forekomsttype som kan defineres som konfigurasjonsparametere i arbeidsflyten.

Installasjon og gjennomgang

Detaljerte trinnvise installasjonsinstruksjoner for å distribuere denne løsningen finner du i vår GitHub repository. Vi brukte en Jupyter-notisbok med Python-kodeceller for å sette opp Airflow DAG. Forutsatt at du allerede har generert dataflytfilen, er følgende en oversikt over installasjonstrinnene på høyt nivå:

1. Lag en S3-bøtte og påfølgende mapper som kreves av Amazon MWAA.
2. Lag et Amazon MWAA-miljø. Merk at vi brukte Airflow versjon 2.0.2 for denne løsningen.
3. Opprett og last opp en requirements.txt fil med alle Python-avhengighetene som kreves av Airflow-oppgavene og last den opp til /requirements katalogen i Amazons MWAA primære S3-bøtte. Dette brukes av det administrerte Airflow-miljøet for å installere Python-avhengighetene.
4. Last opp SMDataWranglerOperator.py filen til /dags katalog. Dette Python-skriptet inneholder kode for den tilpassede Airflow-operatøren for Data Wrangler. Denne operatoren kan brukes til oppgaver for å behandle enhver .flow-fil.
5. Opprett og last opp config.py skript til /dags katalog. Dette Python-skriptet brukes for det første trinnet i vår DAG for å lage konfigurasjonsobjekter som kreves av de resterende trinnene i arbeidsflyten.
6. Til slutt oppretter og laster du opp ml_pipelines.py filen til /dags katalog. Dette skriptet inneholder DAG-definisjonen for Airflow-arbeidsflyten. Det er her vi definerer hver av oppgavene, og setter opp avhengigheter mellom dem. Amazon MWAA måler med jevne mellomrom /dags katalogen for å kjøre dette skriptet for å opprette DAG eller oppdatere den eksisterende med eventuelle siste endringer.

Følgende er koden for SageMaker_DataWrangler_step, som bruker skikken SageMakerDataWranglerOperator. Med bare noen få linjer med kode i DAG-definisjonen Python-skript, kan du peke på SageMakerDataWranglerOperator til Data Wrangler-flytfilplasseringen (som er en S3-plassering). Bak kulissene bruker denne operatøren SageMaker Processing kontaktinformasjon management for å behandle .flow-filen for å bruke de definerte transformasjonene på rådatafilene dine. Du kan også spesifisere type instans og antall forekomster som trengs av Data Wrangler-behandlingsjobben.

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

De config parameter godtar en ordbok (nøkkelverdi-par) med tilleggskonfigurasjoner som kreves av behandlingsjobben, for eksempel utdataprefikset til den endelige utdatafilen, type utdatafil (CSV eller Parkett) og URI for den innebygde Data Wrangler-beholderen bilde. Følgende kode er hva config ordbok for SageMakerDataWranglerOperator ser ut som. Disse konfigurasjonene kreves for en SageMaker-behandling prosessor. For detaljer om hver av disse konfigurasjonsparametrene, 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
	}
}

Rydd opp

For å unngå fremtidige kostnader, slett ressursene som er opprettet for løsningene du implementerte.

1. Følg disse instruksjoner gitt i GitHub-depotet for å rydde opp i ressurser opprettet av SageMaker Pipelines-løsningen.
2. Følg disse instruksjoner gitt i GitHub-depotet for å rydde opp i ressurser opprettet av Step Functions-løsningen.
3. Følg disse instruksjoner gitt i GitHub-depotet for å rydde opp i ressurser opprettet av Amazon MWAA-løsningen.

konklusjonen

Dette innlegget demonstrerte hvordan du enkelt kan integrere Data Wrangler med noen av de velkjente arbeidsflytautomatiserings- og orkestreringsteknologiene i AWS. Vi gjennomgikk først et eksempel på brukstilfelle og arkitektur for løsningen som bruker Data Wrangler for dataforbehandling. Vi demonstrerte deretter hvordan du kan integrere Data Wrangler med Pipelines, Step Functions og Amazon MWAA.

Som et neste trinn kan du finne og prøve ut kodeeksemplene og notatbøkene i vår GitHub repository ved å bruke de detaljerte instruksjonene for hver av løsningene som er diskutert i dette innlegget. For å lære mer om hvordan Data Wrangler kan hjelpe ML-arbeidsbelastningene dine, besøk Data Wrangler produktside og Forbered ML-data med Amazon SageMaker Data Wrangler.

Om forfatterne

Rodrigo Alarcon er en senior ML Strategy Solutions Architect med AWS basert i Santiago, Chile. I sin rolle hjelper Rodrigo selskaper av forskjellige størrelser med å generere forretningsresultater gjennom skybaserte AI- og ML-løsninger. Hans interesser inkluderer maskinlæring og cybersikkerhet.

Ganapathi Krishnamoorthi er senior ML Solutions Architect hos AWS. Ganapathi gir foreskrivende veiledning til oppstarts- og bedriftskunder og hjelper dem med å designe og distribuere skyapplikasjoner i stor skala. Han er spesialisert i maskinlæring og er fokusert på å hjelpe kunder med å utnytte AI/ML for deres forretningsresultater. Når han ikke er på jobb, liker han å utforske utendørs og høre på musikk.

Anjan Biswas er en senior AI Services Solutions Architect med fokus på AI/ML og Data Analytics. Anjan er en del av det verdensomspennende AI-tjenesteteamet og jobber med kunder for å hjelpe dem med å forstå og utvikle løsninger på forretningsproblemer med AI og ML. Anjan har over 14 års erfaring med å jobbe med globale forsyningskjeder, produksjons- og detaljhandelsorganisasjoner og hjelper aktivt kunder med å komme i gang og skalere på AWS AI-tjenester.