Minska kostnader och utvecklingstid med Amazon SageMaker Pipelines lokalt läge

Att skapa robusta och återanvändbara pipelines för maskininlärning (ML) kan vara en komplex och tidskrävande process. Utvecklare testar vanligtvis sina bearbetnings- och träningsskript lokalt, men själva pipelinesna testas vanligtvis i molnet. Att skapa och köra en fullständig pipeline under experimentering lägger till oönskade omkostnader och kostnader till utvecklingens livscykel. I det här inlägget beskriver vi hur du kan använda Amazon SageMaker Pipelines lokalt läge att köra ML pipelines lokalt för att minska både pipelineutveckling och körtid samtidigt som kostnaden minskar. Efter att pipelinen har testats helt lokalt kan du enkelt köra den igen med Amazon SageMaker hanterade resurser med bara några rader kodändringar.

Översikt över ML:s livscykel

En av de viktigaste drivkrafterna för nya innovationer och applikationer i ML är tillgängligheten och mängden data tillsammans med billigare beräkningsalternativ. Inom flera domäner har ML visat sig kapabel att lösa problem som tidigare varit olösliga med klassisk big data och analytiska tekniker, och efterfrågan på datavetenskap och ML-utövare ökar stadigt. Från en mycket hög nivå består ML-livscykeln av många olika delar, men byggandet av en ML-modell består vanligtvis av följande generella steg:

1. Datarensning och förberedelse (funktionsteknik)
2. Modellträning och trimning
3. Modellutvärdering
4. Modellimplementering (eller batchtransformering)

I databeredningssteget laddas data, masseras och omvandlas till den typ av indata eller funktioner som ML-modellen förväntar sig. Att skriva skript för att transformera data är vanligtvis en iterativ process, där snabba återkopplingsslingor är viktiga för att påskynda utvecklingen. Det är normalt inte nödvändigt att använda hela datasetet när du testar funktionsteknikskript, varför du kan använda lokallägesfunktion av SageMaker Processing. Detta gör att du kan köra lokalt och uppdatera koden iterativt med hjälp av en mindre datauppsättning. När den slutliga koden är klar skickas den till fjärrbearbetningsjobbet, som använder hela datasetet och körs på SageMaker-hanterade instanser.

Utvecklingsprocessen liknar dataförberedelsesteget för både modellutbildning och modellutvärderingssteg. Dataforskare använder lokallägesfunktion av SageMaker Training för att snabbt iterera med mindre datauppsättningar lokalt, innan du använder all data i ett SageMaker-hanterat kluster av ML-optimerade instanser. Detta påskyndar utvecklingsprocessen och eliminerar kostnaden för att köra ML-instanser som hanteras av SageMaker medan du experimenterar.

När en organisations ML-mognad ökar kan du använda Amazon SageMaker-rörledningar att skapa ML-pipelines som sammanfogar dessa steg, vilket skapar mer komplexa ML-arbetsflöden som bearbetar, tränar och utvärderar ML-modeller. SageMaker Pipelines är en helt hanterad tjänst för att automatisera de olika stegen i ML-arbetsflödet, inklusive dataladdning, datatransformation, modellträning och inställning samt modelldistribution. Tills nyligen kunde du utveckla och testa dina skript lokalt men var tvungen att testa dina ML-pipelines i molnet. Detta gjorde att iterationen av ML-pipelines flöde och form blev en långsam och kostsam process. Nu, med den extra lokallägesfunktionen i SageMaker Pipelines, kan du iterera och testa dina ML-pipelines på samma sätt som du testar och itererar på dina bearbetnings- och träningsskript. Du kan köra och testa dina pipelines på din lokala dator genom att använda en liten delmängd av data för att validera pipelinesyntaxen och funktionerna.

SageMaker-rörledningar

SageMaker Pipelines tillhandahåller ett helautomatiskt sätt att köra enkla eller komplexa ML-arbetsflöden. Med SageMaker Pipelines kan du skapa ML-arbetsflöden med en lättanvänd Python SDK och sedan visualisera och hantera ditt arbetsflöde med hjälp av Amazon SageMaker Studio. Dina datavetenskapsteam kan bli mer effektiva och skala snabbare genom att lagra och återanvända arbetsflödesstegen du skapar i SageMaker Pipelines. Du kan också använda förbyggda mallar som automatiserar skapandet av infrastruktur och arkiv för att bygga, testa, registrera och distribuera modeller i din ML-miljö. Dessa mallar är automatiskt tillgängliga för din organisation och tillhandahålls med hjälp av AWS servicekatalog produkter.

SageMaker Pipelines ger kontinuerlig integration och kontinuerlig implementering (CI/CD) till ML, som att upprätthålla paritet mellan utvecklings- och produktionsmiljöer, versionskontroll, on-demand-testning och end-to-end-automatisering, vilket hjälper dig att skala ML genom hela din organisation. DevOps-utövare vet att några av de främsta fördelarna med att använda CI/CD-tekniker inkluderar en ökning av produktiviteten via återanvändbara komponenter och en ökning av kvaliteten genom automatiserade tester, vilket leder till snabbare avkastning på investeringen för dina affärsmål. Dessa fördelar är nu tillgängliga för MLOps-utövare genom att använda SageMaker Pipelines för att automatisera träning, testning och distribution av ML-modeller. Med lokalt läge kan du nu iterera mycket snabbare samtidigt som du utvecklar skript för användning i en pipeline. Observera att lokala pipeline-instanser inte kan ses eller köras inom Studio IDE; ytterligare visningsalternativ för lokala pipelines kommer dock att finnas tillgängliga snart.

SageMaker SDK tillhandahåller ett allmänt syfte lokal lägeskonfiguration som gör det möjligt för utvecklare att köra och testa processorer och estimatorer som stöds i sin lokala miljö. Du kan använda träning i lokalt läge med flera AWS-stödda ramverksbilder (TensorFlow, MXNet, Chainer, PyTorch och Scikit-Learn) såväl som bilder du själv tillhandahåller.

SageMaker Pipelines, som bygger en Directed Acyclic Graph (DAG) av orkestrerade arbetsflödessteg, stöder många aktiviteter som är en del av ML-livscykeln. I lokalt läge stöds följande steg:

• Bearbetar arbetssteg – En förenklad, hanterad upplevelse på SageMaker för att köra arbetsbelastningar för databearbetning, såsom funktionsteknik, datavalidering, modellutvärdering och modelltolkning
• Utbildning arbetssteg – En iterativ process som lär en modell att göra förutsägelser genom att presentera exempel från en träningsdatauppsättning
• Hyperparameter tuning jobb – Ett automatiserat sätt att utvärdera och välja de hyperparametrar som ger den mest exakta modellen
• Villkorliga löpsteg – Ett steg som ger en villkorad körning av grenar i en pipeline
• Modellsteg – Med hjälp av CreateModel-argument kan detta steg skapa en modell för användning i transformeringssteg eller senare distribution som en slutpunkt
• Förvandla arbetssteg – Ett batchtransformeringsjobb som genererar förutsägelser från stora datamängder och kör slutledning när en beständig slutpunkt inte behövs
• Misslyckade steg – Ett steg som stoppar en pipelinekörning och markerar körningen som misslyckad

Lösningsöversikt

Vår lösning visar de väsentliga stegen för att skapa och köra SageMaker Pipelines i lokalt läge, vilket innebär att man använder lokala CPU-, RAM- och diskresurser för att ladda och köra arbetsflödesstegen. Din lokala miljö kan köras på en bärbar dator med populära IDE:er som VSCode eller PyCharm, eller så kan den vara värd hos SageMaker med klassiska anteckningsbokinstanser.

Lokalt läge tillåter datavetare att sy ihop steg, som kan inkludera bearbetnings-, utbildnings- och utvärderingsjobb, och köra hela arbetsflödet lokalt. När du är klar med att testa lokalt kan du köra om pipelinen i en SageMaker-hanterad miljö genom att ersätta LocalPipelineSession objekt med PipelineSession, vilket ger konsekvens till ML-livscykeln.

För det här anteckningsbokexemplet använder vi en allmänt tillgänglig standarduppsättning, den UCI Machine Learning Abalone Dataset. Målet är att träna en ML-modell för att bestämma åldern på en fjällsnigel utifrån dess fysiska mätningar. I kärnan är detta ett regressionsproblem.

All kod som krävs för att köra detta anteckningsboksexempel är tillgänglig på GitHub i amazon-sagemaker-exempel förvaret. I det här anteckningsbokexemplet skapas varje pipeline-arbetsflödessteg oberoende och kopplas sedan samman för att skapa pipeline. Vi skapar följande steg:

• Bearbetningssteg (funktionsteknik)
• Träningssteg (modellträning)
• Bearbetningssteg (modellutvärdering)
• Villkorssteg (modellnoggrannhet)
• Skapa modellsteg (modell)
• Transformeringssteg (batchtransformering)
• Registrera modellsteg (modellpaket)
• Felsteg (körning misslyckades)

Följande diagram illustrerar vår pipeline.

Förutsättningar

För att följa med i det här inlägget behöver du följande:

När dessa förutsättningar är på plats kan du köra exempelanteckningsboken enligt beskrivningen i följande avsnitt.

Bygg din pipeline

I det här anteckningsbokexemplet använder vi SageMaker skriptläge för de flesta av ML-processerna, vilket innebär att vi tillhandahåller den faktiska Python-koden (skript) för att utföra aktiviteten och skicka en referens till denna kod. Skriptläge ger stor flexibilitet för att kontrollera beteendet inom SageMaker-bearbetningen genom att låta dig anpassa din kod samtidigt som du drar fördel av SageMaker förbyggda behållare som XGBoost eller Scikit-Learn. Den anpassade koden skrivs till en Python-skriptfil med hjälp av celler som börjar med det magiska kommandot %%writefile, som följande:

%%writefile code/evaluation.py

Den primära aktiveraren för lokalt läge är LocalPipelineSession objekt, som är instansierat från Python SDK. Följande kodsegment visar hur man skapar en SageMaker-pipeline i lokalt läge. Även om du kan konfigurera en lokal datasökväg för många av de lokala pipeline-stegen, är Amazon S3 standardplatsen för att lagra datautmatningen av transformationen. Den nya LocalPipelineSession objekt skickas till Python SDK i många av SageMakers arbetsflödes API-anrop som beskrivs i det här inlägget. Observera att du kan använda local_pipeline_session variabel för att hämta referenser till S3-standardbucket och det aktuella regionnamnet.

from sagemaker.workflow.pipeline_context import LocalPipelineSession

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

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

Innan vi skapar de individuella pipeline-stegen ställer vi in ​​några parametrar som används av pipeline. Vissa av dessa parametrar är strängliterala, medan andra skapas som speciella uppräknade typer som tillhandahålls av SDK. Den uppräknade skrivningen säkerställer att giltiga inställningar tillhandahålls till pipelinen, som den här, som skickas till ConditionLessThanOrEqualTo steg längre ner:

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

För att skapa ett databearbetningssteg, som här används för att utföra funktionsutveckling, använder vi SKLearnProcessor för att ladda och transformera datamängden. Vi passerar local_pipeline_session variabel till klasskonstruktorn, som instruerar arbetsflödessteget att köras i lokalt läge:

from sagemaker.sklearn.processing import SKLearnProcessor

framework_version = "1.0-1"

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

Därefter skapar vi vårt första faktiska pipelinesteg, a ProcessingStep objekt, som importerats från SageMaker SDK. Processorargumenten returneras från ett anrop till SKLearnProcessor run() metod. Detta arbetsflödessteg kombineras med andra steg mot slutet av notebook-datorn för att indikera operationsordningen inom pipeline.

from sagemaker.processing import ProcessingInput, ProcessingOutput
from sagemaker.workflow.steps import ProcessingStep

processor_args = sklearn_processor.run(
    inputs=[
        ProcessingInput(source=input_data, destination="/opt/ml/processing/input"),
    ],
    outputs=[
        ProcessingOutput(output_name="train", source="/opt/ml/processing/train"),
        ProcessingOutput(output_name="validation", source="/opt/ml/processing/validation"),
        ProcessingOutput(output_name="test", source="/opt/ml/processing/test"),
    ],
    code="code/preprocessing.py",
)

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

Därefter tillhandahåller vi kod för att upprätta ett träningssteg genom att först instansiera en standardkalkylator med SageMaker SDK. Vi passerar detsamma local_pipeline_session variabel till estimatorn, som heter xgb_train, som sagemaker_session argument. Eftersom vi vill träna en XGBoost-modell måste vi generera en giltig bild-URI genom att ange följande parametrar, inklusive ramverket och flera versionsparametrar:

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

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

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

Vi kan valfritt kalla ytterligare estimatormetoder, till exempel set_hyperparameters(), för att tillhandahålla hyperparameterinställningar för träningsjobbet. Nu när vi har en estimator konfigurerad är vi redo att skapa själva utbildningssteget. Återigen importerar vi TrainingStep klass från SageMaker SDK-biblioteket:

from sagemaker.workflow.steps import TrainingStep

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

Därefter bygger vi ytterligare ett bearbetningssteg för att utföra modellutvärdering. Detta görs genom att skapa en ScriptProcessor instans och passerar local_pipeline_session objekt som en parameter:

from sagemaker.processing import ScriptProcessor

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

För att möjliggöra distribution av den tränade modellen, antingen till en SageMaker realtidsslutpunkt eller till en batchtransform, måste vi skapa en Model objekt genom att skicka modellartefakter, rätt bild-URI och eventuellt vår anpassade slutledningskod. Vi passerar sedan detta Model invända mot en ModelStep, som läggs till den lokala pipelinen. Se följande kod:

from sagemaker.model import Model

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

from sagemaker.workflow.model_step import ModelStep

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

Därefter skapar vi ett batchtransformeringssteg där vi skickar in en uppsättning funktionsvektorer och utför slutledning. Vi måste först skapa en Transformer objekt och passera local_pipeline_session parameter till den. Sedan skapar vi en TransformStep, skicka de nödvändiga argumenten och lägg till detta till pipelinedefinitionen:

from sagemaker.transformer import Transformer

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

from sagemaker.workflow.steps import TransformStep

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

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

Slutligen vill vi lägga till ett grenvillkor i arbetsflödet så att vi bara kör batchtransform om resultaten av modellutvärderingen uppfyller våra kriterier. Vi kan indikera detta villkor genom att lägga till a ConditionStep med en speciell tillståndstyp, som ConditionLessThanOrEqualTo. Vi räknar sedan upp stegen för de två grenarna, och definierar i huvudsak om/annat eller sant/falskt grenar av pipelinen. If_steps som finns i ConditionStep (steg_skapa_modell, step_transform) körs närhelst villkoret utvärderas till True.

from sagemaker.workflow.conditions import ConditionLessThanOrEqualTo
from sagemaker.workflow.condition_step import ConditionStep
from sagemaker.workflow.functions import JsonGet

cond_lte = ConditionLessThanOrEqualTo(
    left=JsonGet(
        step_name=step_eval.name,
        property_file=evaluation_report,
        json_path="regression_metrics.mse.value",),
    right=mse_threshold,
)

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

Följande diagram illustrerar denna villkorliga gren och de associerade if/else-steg. Endast en gren körs, baserat på resultatet av modellutvärderingssteget jämfört med tillståndssteget.

Nu när vi har alla våra steg definierade och de underliggande klassinstanserna skapade, kan vi kombinera dem till en pipeline. Vi tillhandahåller några parametrar och definierar på ett avgörande sätt operationsordningen genom att helt enkelt lista stegen i önskad ordning. Observera att TransformStep visas inte här eftersom det är målet för det villkorliga steget och tillhandahölls som stegargument till ConditionalStep tidigare.

from sagemaker.workflow.pipeline import Pipeline

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

För att köra pipelinen måste du anropa två metoder: pipeline.upsert(), som laddar upp pipelinen till den underliggande tjänsten, och pipeline.start(), som börjar köra pipelinen. Du kan använda olika andra metoder för att undersöka körstatus, lista pipelinestegen och mer. Eftersom vi använde pipeline-sessionen för lokalt läge, körs alla dessa steg lokalt på din processor. Cellutgången under startmetoden visar utdata från pipeline:

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

Du bör se ett meddelande längst ned i cellutgången som liknar följande:

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

Återgå till hanterade resurser

Efter att vi har bekräftat att pipelinen körs utan fel och vi är nöjda med flödet och formen på pipelinen, kan vi återskapa pipelinen men med SageMaker-hanterade resurser och köra den igen. Den enda ändring som krävs är att använda PipelineSession objekt istället för LocalPipelineSession:

från sagemaker.workflow.pipeline_context importera LocalPipelineSession
from sagemaker.workflow.pipeline_context import PipelineSession

local_pipeline_session = LocalPipelineSession()
pipeline_session = PipelineSession()

Detta informerar tjänsten om att köra varje steg som refererar till detta sessionsobjekt på SageMaker-hanterade resurser. Med tanke på den lilla ändringen illustrerar vi endast de nödvändiga kodändringarna i följande kodcell, men samma ändring skulle behöva implementeras på varje cell med hjälp av local_pipeline_session objekt. Ändringarna är dock identiska i alla celler eftersom vi bara ersätter local_pipeline_session objekt med pipeline_session objekt.

from sagemaker.sklearn.processing import SKLearnProcessor

framework_version = "1.0-1"

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

Efter att det lokala sessionsobjektet har ersatts överallt, återskapar vi pipelinen och kör den med SageMaker-hanterade resurser:

from sagemaker.workflow.pipeline import Pipeline

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

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

Städa upp

Om du vill hålla ordning på Studiomiljön kan du använda följande metoder för att ta bort SageMaker-pipelinen och modellen. Hela koden finns i exemplet anteckningsbok.

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

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

Slutsats

Tills nyligen kunde du använda lokallägesfunktionen i SageMaker Processing och SageMaker Training för att iterera på dina bearbetnings- och träningsskript lokalt, innan du körde dem på all data med SageMaker-hanterade resurser. Med den nya funktionen för lokalt läge i SageMaker Pipelines kan ML-utövare nu tillämpa samma metod när de itererar på sina ML-pipelines, och sätter ihop de olika ML-arbetsflödena. När pipelinen är klar för produktion krävs bara några rader kodändringar för att köra den med SageMaker-hanterade resurser. Detta minskar pipelinekörningstiden under utveckling, vilket leder till snabbare pipelineutveckling med snabbare utvecklingscykler, samtidigt som kostnaden för SageMaker-hanterade resurser minskar.

Om du vill veta mer, besök Amazon SageMaker-rörledningar or Använd SageMaker Pipelines för att köra dina jobb lokalt.

Om författarna

Paul Hargis har fokuserat sina ansträngningar på maskininlärning på flera företag, inklusive AWS, Amazon och Hortonworks. Han tycker om att bygga tekniska lösningar och lära människor hur man får ut det mesta av det. Innan han började på AWS var han ledande arkitekt för Amazon Exports and Expansions, och hjälpte amazon.com att förbättra upplevelsen för internationella shoppare. Paul gillar att hjälpa kunder att utöka sina maskininlärningsinitiativ för att lösa verkliga problem.

Niklas Palm är lösningsarkitekt på AWS i Stockholm, där han hjälper kunder över hela Norden att lyckas i molnet. Han brinner särskilt för serverlös teknik tillsammans med IoT och maskininlärning. Utanför arbetet är Niklas en ivrig längdskidåkare och snowboardåkare samt en mästare.

Kirit Thadaka är en ML Solutions Architect som arbetar i SageMaker Service SA-teamet. Innan Kirit började på AWS arbetade Kirit med AI-startups i ett tidigt skede, följt av en tid konsultation i olika roller inom AI-forskning, MLOps och tekniskt ledarskap.