Integrera Amazon SageMaker Data Wrangler med MLOps arbetsflöden

När företag går från att köra ad hoc-modeller för maskininlärning (ML) till att använda AI/ML för att transformera sin verksamhet i stor skala, blir införandet av ML Operations (MLOps) oundvikligt. Som visas i följande figur börjar ML-livscykeln med att skapa ett affärsproblem som ett ML-användningsfall följt av en serie faser, inklusive dataförberedelse, funktionsutveckling, modellbyggande, implementering, kontinuerlig övervakning och omskolning. För många företag är många av dessa steg fortfarande manuella och löst integrerade med varandra. Därför är det viktigt att automatisera hela ML-livscykeln, vilket möjliggör frekventa experiment för att skapa bättre affärsresultat. Dataförberedelse är ett av de avgörande stegen i denna livscykel, eftersom ML-modellens noggrannhet beror på kvaliteten på träningsdatauppsättningen.

Dataforskare och ML-ingenjörer spenderar någonstans mellan 70–80 % av sin tid på att samla in, analysera, rengöra och transformera data som krävs för modellutbildning. Amazon SageMaker Data Wrangler är en fullt hanterad förmåga Amazon SageMaker som gör det snabbare för datavetare och ML-ingenjörer att analysera och förbereda data för sina ML-projekt med lite eller ingen kod. När det kommer till operationalisering av en end-to-end ML-livscykel är dataförberedelse nästan alltid det första steget i processen. Med tanke på att det finns många sätt att bygga en end-to-end ML-pipeline diskuterar vi i det här inlägget hur du enkelt kan integrera Data Wrangler med några av de välkända arbetsflödesautomations- och orkestreringsteknikerna.

Lösningsöversikt

I det här inlägget visar vi hur användare kan integrera databeredning med Data Wrangler med Amazon SageMaker-rörledningar, AWS stegfunktioneroch Apache luftflöde med Amazon Managed Workflow för Apache Airflow (Amazon MWAA). Pipelines är en SageMaker-funktion som är en specialbyggd och lättanvänd tjänst för kontinuerlig integration och kontinuerlig leverans (CI/CD) för ML. Step Functions är en serverlös, visuell arbetsflödestjänst med låg kod som används för att orkestrera AWS-tjänster och automatisera affärsprocesser. Amazon MWAA är en hanterad orkestreringstjänst för Apache Airflow som gör det enklare att driva end-to-end-data och ML-pipelines.

För demonstrationsändamål överväger vi ett användningsfall för att förbereda data för att träna en ML-modell med SageMaker inbyggd XGBoost-algoritm som hjälper oss att identifiera bedrägliga fordonsförsäkringskrav. Vi använde en syntetiskt genererad uppsättning exempeldata för att träna modellen och skapa en SageMaker-modell med hjälp av modellartefakter från träningsprocessen. Vårt mål är att operationalisera denna process från början till slut genom att sätta upp ett ML-arbetsflöde. Även om ML-arbetsflöden kan vara mer utarbetade använder vi ett minimalt arbetsflöde för demonstrationsändamål. Det första steget i arbetsflödet är dataförberedelse med Data Wrangler, följt av ett modellutbildningssteg och slutligen ett steg för att skapa modell. Följande diagram illustrerar vårt lösningsarbetsflöde.

I följande avsnitt går vi igenom hur du ställer in ett Data Wrangler-flöde och integrerar Data Wrangler med Pipelines, Step Functions och Apache Airflow.

Ställ in ett Data Wrangler-flöde

Vi börjar med att skapa ett Data Wrangler-flöde, även kallat a data flöde, använda dataflöde UI via Amazon SageMaker Studio ID. Vår exempeldatauppsättning består av två datafiler: claims.csv och kunder.csv, som lagras i en Amazon enkel lagringstjänst (Amazon S3) skopa. Vi använder dataflödesgränssnittet för att tillämpa Data Wranglers inbyggda transformationer sådan kategorisk kodning, strängformatering och tillskrivning till funktionskolumnerna i var och en av dessa filer. Vi ansöker också anpassad transformation till några funktionskolumner med några rader anpassad Python-kod med Pandas DataFrame. Följande skärmdump visar omvandlingarna som tillämpas på filen claims.csv i dataflödesgränssnittet.

Slutligen sammanfogar vi resultaten av de tillämpade transformationerna av de två datafilerna för att generera en enda träningsdatauppsättning för vår modellträning. Vi använder Data Wranglers inbyggda gå med i datauppsättningar förmåga, som låter oss utföra SQL-liknande joinoperationer på tabelldata. Följande skärmdump visar dataflödet i dataflödesgränssnittet i Studio. För steg-för-steg-instruktioner för att skapa dataflödet med Data Wrangler, se GitHub repository.

Du kan nu använda dataflödesfilen (.flow) för att utföra datatransformationer på våra rådatafiler. Dataflödesgränssnittet kan automatiskt generera Python-anteckningsböcker för oss att använda och integrera direkt med Pipelines med SageMaker SDK. För stegfunktioner använder vi AWS Steg Funktioner Data Science Python SDK att integrera vår Data Wrangler-behandling med en Step Functions-pipeline. För Amazon MWAA använder vi en anpassad luftflödesoperatör och Airflow SageMaker operatör. Vi diskuterar var och en av dessa tillvägagångssätt i detalj i följande avsnitt.

Integrera Data Wrangler med Pipelines

SageMaker Pipelines är ett inbyggt arbetsflödesorkestreringsverktyg för att bygga ML-pipelines som drar fördel av direkt SageMaker-integration. Tillsammans med SageMaker-modellregister, Pipelines förbättrar den operativa motståndskraften och reproducerbarheten för dina ML-arbetsflöden. Dessa komponenter för automatisering av arbetsflöden gör att du enkelt kan skala din förmåga att bygga, träna, testa och distribuera hundratals modeller i produktionen; iterera snabbare; minska fel på grund av manuell orkestrering; och bygga repeterbara mekanismer. Varje steg i pipelinen kan hålla reda på linjen, och mellansteg kan cachelagras för att snabbt köra pipelinen igen. Du kan skapa pipelines med SageMaker Python SDK.

Ett arbetsflöde byggt med SageMaker pipelines består av en sekvens av steg som bildar en Directed Acyclic Graph (DAG). I det här exemplet börjar vi med a bearbetningssteg, som kör en SageMaker Bearbetar jobb baserat på Data Wranglers flödesfil för att skapa en träningsdatauppsättning. Vi fortsätter sedan med en träningssteg, där vi tränar en XGBoost-modell med hjälp av SageMakers inbyggda XGBoost-algoritm och träningsdataset som skapades i föregående steg. Efter att en modell har tränats avslutar vi detta arbetsflöde med en RegisterModel steg för att registrera den tränade modellen med SageMakers modellregister.

Installation och genomgång

För att köra detta exempel använder vi en Jupyter-anteckningsbok som kör Python3 på en Data Science-kärnavbildning i en Studio-miljö. Du kan också köra den på en Jupyter notebook-instans lokalt på din maskin genom att ställa in autentiseringsuppgifterna för att ta på sig SageMaker-exekveringsrollen. Den bärbara datorn är lätt och kan köras på en ml.t3.medium-instans. Detaljerade steg-för-steg-instruktioner finns i GitHub repository.

Du kan antingen använda exportfunktionen i Data Wrangler för att generera Pipelines-koden, eller bygga ditt eget skript från grunden. I vårt provlager använder vi en kombination av båda metoderna för enkelhetens skull. På en hög nivå är följande steg för att bygga och köra Pipelines-arbetsflödet:

1. Generera en flödesfil från Data Wrangler eller använd installationsskriptet för att generera en flödesfil från en förkonfigurerad mall.
2. Skapa ett Amazon enkel lagringstjänst (Amazon S3) hink och ladda upp din flödesfil och indatafiler till hinken. I vår exempelanteckningsbok använder vi SageMaker standard S3-hink.
3. Följ instruktionerna i anteckningsboken för att skapa ett Processor-objekt baserat på Data Wrangler-flödesfilen och ett Estimator-objekt med parametrarna för träningsjobbet.
   1. I vårt exempel, eftersom vi bara använder SageMaker-funktioner och standard S3-bucket, kan vi använda Studios standardexekveringsroll. Det samma AWS identitets- och åtkomsthantering (IAM)-rollen antas av pipelinekörningen, bearbetningsjobbet och utbildningsjobbet. Du kan ytterligare anpassa exekveringsrollen enligt minimibehörighet.
4. Fortsätt med instruktionerna för att skapa en pipeline med steg som refererar till Processor- och Estimator-objekten och kör sedan pipelinen. Bearbetnings- och utbildningsjobben körs i SageMaker-hanterade miljöer och tar några minuter att slutföra.
5. I Studio kan du se pipelinedetaljerna övervaka pipelinekörningen. Du kan också övervaka de underliggande bearbetnings- och utbildningsjobben från SageMaker-konsolen eller från amazoncloudwatch.

Integrera Data Wrangler med stegfunktioner

Med Step Functions kan du uttrycka komplex affärslogik som lågkodade, händelsedrivna arbetsflöden som kopplar samman olika AWS-tjänster. De Steg Funktioner Data Science SDK är ett bibliotek med öppen källkod som tillåter datavetare att skapa arbetsflöden som kan förbehandla datamängder och bygga, distribuera och övervaka ML-modeller med SageMaker och Step Functions. Stegfunktioner bygger på tillståndsmaskiner och uppgifter. Stegfunktioner skapar arbetsflöden utifrån anropade steg tillstånd, och uttrycker det arbetsflödet i Språket i Amazonas stater. När du skapar ett arbetsflöde med Step Functions Data Science SDK skapar det en tillståndsmaskin som representerar ditt arbetsflöde och steg i Step Functions.

För det här användningsfallet byggde vi ett arbetsflöde för stegfunktioner baserat på det vanliga mönstret som används i det här inlägget som inkluderar ett bearbetningssteg, träningssteg och RegisterModel steg. I det här fallet importerar vi dessa steg från Step Functions Data Science Python SDK. Vi kopplar ihop dessa steg i samma ordning för att skapa ett arbetsflöde för Step Functions. Arbetsflödet använder flödesfilen som genererades från Data Wrangler, men du kan också använda din egen Data Wrangler-flödesfil. Vi återanvänder en del kod från Data Wrangler exportfunktion för enkelheten. Vi kör dataförbearbetningslogiken som genereras av Data Wrangler-flödesfilen för att skapa en träningsdatauppsättning, träna en modell med XGBoost-algoritmen och spara den tränade modellartefakten som en SageMaker-modell. Dessutom, i GitHub-repo, visar vi också hur Step Functions tillåter oss att försöka fånga upp fel och hantera fel och återförsök med FailStateStep och CatchStateStep.

Det resulterande flödesdiagrammet, som visas i följande skärmdump, är tillgängligt på Step Functions-konsolen efter att arbetsflödet har startat. Detta hjälper datavetare och ingenjörer att visualisera hela arbetsflödet och varje steg i det, och få tillgång till de länkade CloudWatch-loggarna för varje steg.

Installation och genomgång

För att köra detta exempel använder vi en Python-anteckningsbok som körs med en Data Science-kärna i en Studio-miljö. Du kan också köra den på en Python-anteckningsbok-instans lokalt på din maskin genom att ställa in autentiseringsuppgifterna för att anta SageMaker-exekveringsrollen. Den bärbara datorn är lätt och kan till exempel köras på en t3 medium. Detaljerade steg-för-steg-instruktioner finns i GitHub repository.

Du kan antingen använda exportfunktionen i Data Wrangler för att generera Pipelines-koden och modifiera den för stegfunktioner eller bygga ditt eget skript från grunden. I vårt provlager använder vi en kombination av båda metoderna för enkelhetens skull. På en hög nivå är följande steg för att bygga och köra arbetsflödet för Step Functions:

1. Generera en flödesfil från Data Wrangler eller använd installationsskriptet för att generera en flödesfil från en förkonfigurerad mall.
2. Skapa en S3-bucket och ladda upp din flödesfil och indatafiler till hinken.
3. Konfigurera din SageMaker-exekveringsroll med de nödvändiga behörigheterna som nämnts tidigare. Se GitHub-förvaret för detaljerade instruktioner.
4. Följ instruktionerna för att köra anteckningsboken i arkivet för att starta ett arbetsflöde. Bearbetningsjobbet körs i en SageMaker-hanterad Spark-miljö och kan ta några minuter att slutföra.
5. Gå till Step Functions-konsolen och spåra arbetsflödet visuellt. Du kan också navigera till de länkade CloudWatch-loggarna för att felsöka fel.

Låt oss granska några viktiga delar av koden här. För att definiera arbetsflödet definierar vi först stegen i arbetsflödet för Step Function-tillståndsmaskinen. Det första steget är data_wrangler_step för databehandling, som använder Data Wrangler-flödesfilen som indata för att transformera rådatafilerna. Vi definierar också ett modellutbildningssteg och ett modellskapande steg med namnet training_step och model_step, respektive. Slutligen skapar vi ett arbetsflöde genom att kedja alla steg vi skapade, som visas i följande kod:

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 exempel byggde vi arbetsflödet för att ta jobbnamn som parametrar eftersom de är unika och måste genereras slumpmässigt under varje pipelinekörning. Vi skickar dessa namn när arbetsflödet körs. Du kan också schemalägga arbetsflödet för Step Functions att köras med CloudWatch (se Schemalägg ett serverlöst arbetsflöde med AWS Step Functions och Amazon CloudWatch), anropade med hjälp av Amazon S3-evenemang, eller åberopas från Amazon EventBridge (Se Skapa en EventBridge-regel som utlöser ett Step Functions-arbetsflöde). I demonstrationssyfte kan vi anropa Step Functions-arbetsflödet från Step Functions-konsolens användargränssnitt eller använda följande kod från anteckningsboken.

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

Integrera Data Wrangler med Apache Airflow

Ett annat populärt sätt att skapa ML-arbetsflöden är att använda Apache Airflow. Apache Airflow är en öppen källkodsplattform som låter dig skapa, schemalägga och övervaka arbetsflöden programmatiskt. Amazon MWAA gör det enkelt att sätta upp och driva end-to-end ML-pipelines med Apache Airflow i molnet i stor skala. En luftflödespipeline består av en sekvens av uppgifter, även kallad en arbetsflöde. Ett arbetsflöde definieras som en DAG som kapslar in uppgifterna och beroenden mellan dem och definierar hur de ska köras inom arbetsflödet.

Vi skapade en Airflow DAG i en Amazon MWAA-miljö för att implementera vårt MLOps-arbetsflöde. Varje uppgift i arbetsflödet är en körbar enhet, skriven i programmeringsspråket Python, som utför någon åtgärd. En uppgift kan antingen vara en Operatören eller ett sensor. I vårt fall använder vi en Airflow Python-operatör tillsammans med SageMaker Python SDK för att köra Data Wrangler Python-skriptet och använda Airflow som stöds inbyggt SageMaker-operatörer att träna den inbyggda SageMaker XGBoost-algoritmen och skapa modellen från de resulterande artefakterna. Vi skapade också en användbar anpassad Data Wrangler-operatör (SageMakerDataWranglerOperator) för Apache Airflow, som du kan använda för att bearbeta Data Wrangler-flödesfiler för databehandling utan att behöva någon ytterligare kod.

Följande skärmdump visar Airflow DAG med fem steg för att implementera vårt MLOps-arbetsflöde.

Startsteget använder en Python-operator för att initiera konfigurationer för resten av stegen i arbetsflödet. SageMaker_DataWrangler_Step användningar SageMakerDataWranglerOperator och dataflödesfilen vi skapade tidigare. SageMaker_training_step och SageMaker_create_model_step använd de inbyggda SageMaker-operatörerna för modellträning respektive modellskapande. Vår Amazon MWAA-miljö använder den minsta instanstypen (mw1.small), eftersom huvuddelen av bearbetningen sker via Processing-jobb, som använder sin egen instanstyp som kan definieras som konfigurationsparametrar inom arbetsflödet.

Installation och genomgång

Detaljerade steg-för-steg installationsinstruktioner för att implementera denna lösning finns i vår GitHub repository. Vi använde en Jupyter-anteckningsbok med Python-kodceller för att ställa in Airflow DAG. Förutsatt att du redan har genererat dataflödesfilen är följande en översikt över installationsstegen på hög nivå:

1. Skapa en S3-hink och efterföljande mappar som krävs av Amazon MWAA.
2. Skapa en Amazon MWAA-miljö. Observera att vi använde Airflow version 2.0.2 för denna lösning.
3. Skapa och ladda upp en requirements.txt fil med alla Python-beroenden som krävs av Airflow-uppgifterna och ladda upp den till /requirements katalogen i Amazons MWAA primära S3-hink. Detta används av den hanterade Airflow-miljön för att installera Python-beroenden.
4. Ladda upp SMDataWranglerOperator.py fil till /dags katalog. Detta Python-skript innehåller kod för den anpassade Airflow-operatören för Data Wrangler. Denna operatör kan användas för uppgifter för att bearbeta vilken .flow-fil som helst.
5. Skapa och ladda upp config.py manus till /dags katalog. Detta Python-skript används för det första steget i vår DAG för att skapa konfigurationsobjekt som krävs av de återstående stegen i arbetsflödet.
6. Slutligen, skapa och ladda upp ml_pipelines.py fil till /dags katalog. Det här skriptet innehåller DAG-definitionen för Airflow-arbetsflödet. Det är här vi definierar var och en av uppgifterna och ställer in beroenden mellan dem. Amazon MWAA undersöker regelbundet /dags katalog för att köra det här skriptet för att skapa DAG eller uppdatera den befintliga med de senaste ändringarna.

Följande är koden för SageMaker_DataWrangler_step, som använder seden SageMakerDataWranglerOperator. Med bara några rader kod i ditt DAG-definition Python-skript kan du peka på SageMakerDataWranglerOperator till Data Wrangler-flödesfilplatsen (som är en S3-plats). Bakom kulisserna använder den här operatören SageMaker Processing kontaktinformation förvaltning för att bearbeta .flow-filen för att tillämpa de definierade transformationerna på dina rådatafiler. Du kan också ange typ av instans och antalet instanser som behövs av Data Wrangler-bearbetningsjobbet.

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

Smakämnen config parametern accepterar en ordlista (nyckel-värdepar) med ytterligare konfigurationer som krävs av bearbetningsjobbet, såsom utdataprefixet för den slutliga utdatafilen, typ av utdatafil (CSV eller Parquet) och URI för den inbyggda Data Wrangler-behållaren bild. Följande kod är vad config ordbok för SageMakerDataWranglerOperator ser ut som. Dessa konfigurationer krävs för en SageMaker-bearbetning Processorn. För detaljer om var och en av dessa konfigurationsparametrar, 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
	}
}

Städa upp

Ta bort resurserna som skapats för de lösningar du implementerat för att undvika framtida avgifter.

1. Följ dessa instruktioner tillhandahålls i GitHub-förvaret för att rensa upp resurser skapade av SageMaker Pipelines-lösningen.
2. Följ dessa instruktioner tillhandahålls i GitHub-förvaret för att rensa upp resurser skapade av Step Functions-lösningen.
3. Följ dessa instruktioner tillhandahålls i GitHub-förvaret för att rensa upp resurser skapade av Amazon MWAA-lösningen.

Slutsats

Det här inlägget visade hur du enkelt kan integrera Data Wrangler med några av de välkända arbetsflödesautomations- och orkestreringsteknikerna i AWS. Vi granskade först ett exempel på användningsfall och arkitektur för lösningen som använder Data Wrangler för dataförbehandling. Vi visade sedan hur du kan integrera Data Wrangler med Pipelines, Step Functions och Amazon MWAA.

Som nästa steg kan du hitta och prova kodexemplen och anteckningsböckerna i vår GitHub repository med de detaljerade instruktionerna för var och en av lösningarna som diskuteras i det här inlägget. För att lära dig mer om hur Data Wrangler kan hjälpa dina ML-arbetsbelastningar, besök Data Wrangler Produktsida och Förbered ML-data med Amazon SageMaker Data Wrangler.

Om författarna

Rodrigo Alarcon är en senior ML Strategy Solutions Architect med AWS baserad i Santiago, Chile. I sin roll hjälper Rodrigo företag av olika storlekar att generera affärsresultat genom molnbaserade AI- och ML-lösningar. Hans intressen inkluderar maskininlärning och cybersäkerhet.

Ganapathi Krishnamoorthi är Senior ML Solutions Architect på AWS. Ganapathi tillhandahåller föreskrivande vägledning till startup- och företagskunder och hjälper dem att designa och distribuera molnapplikationer i stor skala. Han är specialiserad på maskininlärning och är fokuserad på att hjälpa kunder att utnyttja AI/ML för sina affärsresultat. När han inte är på jobbet tycker han om att utforska utomhus och lyssna på musik.

Anjan Biswas är en Senior AI Services Solutions Architect med fokus på AI/ML och Data Analytics. Anjan är en del av det världsomspännande AI-serviceteamet och arbetar med kunder för att hjälpa dem att förstå och utveckla lösningar på affärsproblem med AI och ML. Anjan har över 14 års erfarenhet av att arbeta med globala supply chain-, tillverknings- och detaljhandelsorganisationer och hjälper aktivt kunder att komma igång och skala på AWS AI-tjänster.