Reducer omkostninger og udviklingstid med Amazon SageMaker Pipelines lokal tilstand

At skabe robuste og genanvendelige maskinlæringspipelines (ML) kan være en kompleks og tidskrævende proces. Udviklere tester normalt deres behandlings- og træningsscripts lokalt, men selve pipelines testes typisk i skyen. Oprettelse og drift af en fuld pipeline under eksperimentering tilføjer uønskede overhead og omkostninger til udviklingslivscyklussen. I dette indlæg beskriver vi, hvordan du kan bruge Amazon SageMaker Pipelines lokal tilstand at køre ML-pipelines lokalt for at reducere både pipeline-udvikling og driftstid og samtidig reducere omkostningerne. Efter at pipelinen er blevet fuldt testet lokalt, kan du nemt køre den igen med Amazon SageMaker administrerede ressourcer med blot nogle få linjers kodeændringer.

Oversigt over ML livscyklus

En af hoveddriverne for nye innovationer og applikationer i ML er tilgængeligheden og mængden af ​​data sammen med billigere beregningsmuligheder. På flere domæner har ML vist sig at være i stand til at løse problemer, der tidligere var uløselige med klassiske big data og analytiske teknikker, og efterspørgslen efter datavidenskab og ML-praktikere stiger støt. Fra et meget højt niveau består ML-livscyklussen af ​​mange forskellige dele, men opbygningen af ​​en ML-model består normalt af følgende generelle trin:

1. Datarensning og forberedelse (feature engineering)
2. Modeltræning og tuning
3. Modelvurdering
4. Modelimplementering (eller batchtransformation)

I dataforberedelsestrinnet indlæses data, masseres og transformeres til den type input eller funktioner, som ML-modellen forventer. At skrive scripts for at transformere dataene er typisk en iterativ proces, hvor hurtige feedback-loops er vigtige for at fremskynde udviklingen. Det er normalt ikke nødvendigt at bruge det fulde datasæt, når du tester feature engineering scripts, hvorfor du kan bruge lokal tilstand funktion af SageMaker Processing. Dette giver dig mulighed for at køre lokalt og opdatere koden iterativt ved hjælp af et mindre datasæt. Når den endelige kode er klar, sendes den til fjernbehandlingsjobbet, som bruger det komplette datasæt og kører på SageMaker-administrerede forekomster.

Udviklingsprocessen ligner dataforberedelsestrinnet for både modeltræning og modelevalueringstrin. Data videnskabsmænd bruger lokal tilstand funktion af SageMaker Training til at iterere hurtigt med mindre datasæt lokalt, før du bruger alle data i en SageMaker-administreret klynge af ML-optimerede forekomster. Dette fremskynder udviklingsprocessen og eliminerer omkostningerne ved at køre ML-instanser administreret af SageMaker, mens du eksperimenterer.

Efterhånden som en organisations ML-modenhed øges, kan du bruge Amazon SageMaker Pipelines at skabe ML-pipelines, der syr disse trin sammen, hvilket skaber mere komplekse ML-arbejdsgange, der behandler, træner og evaluerer ML-modeller. SageMaker Pipelines er en fuldt administreret service til automatisering af de forskellige trin i ML-workflowet, herunder dataindlæsning, datatransformation, modeltræning og -justering og modelimplementering. Indtil for nylig kunne du udvikle og teste dine scripts lokalt, men skulle teste dine ML-pipelines i skyen. Dette gjorde iteration af flow og form af ML-rørledninger til en langsom og bekostelig proces. Nu, med den tilføjede lokaltilstandsfunktion i SageMaker Pipelines, kan du iterere og teste dine ML-pipelines på samme måde, som du tester og itererer på dine behandlings- og træningsscripts. Du kan køre og teste dine pipelines på din lokale maskine ved at bruge et lille undersæt af data til at validere pipeline-syntaksen og funktionaliteterne.

SageMaker Pipelines

SageMaker Pipelines giver en fuldautomatisk måde at køre simple eller komplekse ML-arbejdsgange på. Med SageMaker Pipelines kan du oprette ML-arbejdsgange med et brugervenligt Python SDK og derefter visualisere og administrere dit arbejdsflow vha. Amazon SageMaker Studio. Dine datavidenskabsteams kan være mere effektive og skalere hurtigere ved at gemme og genbruge de workflow-trin, du opretter i SageMaker Pipelines. Du kan også bruge forudbyggede skabeloner, der automatiserer oprettelsen af ​​infrastruktur og lager til at bygge, teste, registrere og implementere modeller i dit ML-miljø. Disse skabeloner er automatisk tilgængelige for din organisation og klargøres ved hjælp af AWS servicekatalog produkter.

SageMaker Pipelines bringer kontinuerlig integration og kontinuerlig udrulning (CI/CD)-praksis til ML, såsom opretholdelse af paritet mellem udviklings- og produktionsmiljøer, versionskontrol, on-demand-test og end-to-end automatisering, som hjælper dig med at skalere ML gennem hele dit organisation. DevOps-udøvere ved, at nogle af de vigtigste fordele ved at bruge CI/CD-teknikker inkluderer en stigning i produktiviteten via genanvendelige komponenter og en stigning i kvaliteten gennem automatiseret test, hvilket fører til hurtigere ROI for dine forretningsmål. Disse fordele er nu tilgængelige for MLOps-udøvere ved at bruge SageMaker Pipelines til at automatisere træning, test og implementering af ML-modeller. Med lokal tilstand kan du nu iterere meget hurtigere, mens du udvikler scripts til brug i en pipeline. Bemærk, at lokale pipeline-instanser ikke kan ses eller køres i Studio IDE; yderligere visningsmuligheder for lokale rørledninger vil dog snart være tilgængelige.

SageMaker SDK giver et generelt formål lokal tilstandskonfiguration der giver udviklere mulighed for at køre og teste understøttede processorer og estimatorer i deres lokale miljø. Du kan bruge træning i lokal tilstand med flere AWS-understøttede rammebilleder (TensorFlow, MXNet, Chainer, PyTorch og Scikit-Learn) samt billeder, du selv leverer.

SageMaker Pipelines, som bygger en Directed Acyclic Graph (DAG) af orkestrerede workflow-trin, understøtter mange aktiviteter, der er en del af ML-livscyklussen. I lokal tilstand understøttes følgende trin:

• Behandling af jobtrin – En forenklet, administreret oplevelse på SageMaker til at køre databehandlingsarbejdsbelastninger, såsom feature engineering, datavalidering, modelevaluering og modelfortolkning
• Træning af jobtrin – En iterativ proces, der lærer en model at lave forudsigelser ved at præsentere eksempler fra et træningsdatasæt
• Hyperparameter tuning job – En automatiseret måde at evaluere og vælge de hyperparametre, der producerer den mest nøjagtige model
• Betingede løbetrin – Et trin, der giver et betinget løb af grene i en rørledning
• Model trin – Ved hjælp af CreateModel-argumenter kan dette trin oprette en model til brug i transformationstrin eller senere implementering som et slutpunkt
• Forvandle arbejdstrin – Et batch-transformationsjob, der genererer forudsigelser fra store datasæt og kører inferens, når et vedvarende slutpunkt ikke er nødvendigt
• Fejltrin – Et trin, der stopper en pipelinekørsel og markerer kørslen som mislykket

Løsningsoversigt

Vores løsning demonstrerer de væsentlige trin til at skabe og køre SageMaker Pipelines i lokal tilstand, hvilket betyder at bruge lokale CPU-, RAM- og diskressourcer til at indlæse og køre workflow-trinene. Dit lokale miljø kan køre på en bærbar computer ved hjælp af populære IDE'er som VSCode eller PyCharm, eller det kan hostes af SageMaker ved hjælp af klassiske notebook-forekomster.

Lokal tilstand giver datavidenskabsfolk mulighed for at sammenføje trin, som kan omfatte behandlings-, trænings- og evalueringsjob og køre hele arbejdsgangen lokalt. Når du er færdig med at teste lokalt, kan du køre pipelinen igen i et SageMaker-administreret miljø ved at erstatte LocalPipelineSession objekt med PipelineSession, hvilket bringer konsistens til ML-livscyklussen.

Til dette notesbogseksempel bruger vi et standard offentligt tilgængeligt datasæt, den UCI Machine Learning Abalone Datasæt. Målet er at træne en ML-model til at bestemme alderen på en abalonesnegl ud fra dens fysiske mål. I bund og grund er dette et regressionsproblem.

Al den kode, der kræves for at køre denne notebook-eksempel, er tilgængelig på GitHub i amazon-sagemaker-eksempler depot. I denne notesbogseksempel oprettes hvert pipeline-workflowtrin uafhængigt og forbindes derefter sammen for at skabe pipelinen. Vi opretter følgende trin:

• Bearbejdningstrin (feature engineering)
• Træningstrin (modeltræning)
• Bearbejdningstrin (modelevaluering)
• Tilstandstrin (modelnøjagtighed)
• Opret modeltrin (model)
• Transformeringstrin (batchtransformation)
• Registrer modeltrin (modelpakke)
• Fejltrin (kørsel mislykkedes)

Følgende diagram illustrerer vores pipeline.

Forudsætninger

For at følge med i dette indlæg skal du bruge følgende:

Når disse forudsætninger er på plads, kan du køre prøvenotesbogen som beskrevet i de følgende afsnit.

Byg din pipeline

I dette notesbogseksempel bruger vi SageMaker Script-tilstand for de fleste af ML-processerne, hvilket betyder, at vi leverer den faktiske Python-kode (scripts) til at udføre aktiviteten og videregive en reference til denne kode. Script-tilstand giver stor fleksibilitet til at kontrollere adfærden i SageMaker-behandlingen ved at give dig mulighed for at tilpasse din kode, mens du stadig drager fordel af SageMaker forudbyggede containere som XGBoost eller Scikit-Learn. Den brugerdefinerede kode skrives til en Python-scriptfil ved hjælp af celler, der begynder med den magiske kommando %%writefile, som følgende:

%%writefile code/evaluation.py

Den primære aktivering af lokal tilstand er LocalPipelineSession objekt, som er instansieret fra Python SDK. De følgende kodesegmenter viser, hvordan man opretter en SageMaker-pipeline i lokal tilstand. Selvom du kan konfigurere en lokal datasti for mange af de lokale pipeline-trin, er Amazon S3 standardplaceringen til at gemme dataoutput ved transformationen. Den nye LocalPipelineSession objekt sendes til Python SDK i mange af SageMaker workflow API-kaldene beskrevet i dette indlæg. Bemærk, at du kan bruge local_pipeline_session variabel for at hente referencer til S3 standard bucket og det aktuelle regions navn.

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"

Inden vi opretter de individuelle pipeline-trin, sætter vi nogle parametre, der bruges af pipelinen. Nogle af disse parametre er strenge bogstaver, mens andre er oprettet som specielle opregnede typer leveret af SDK. Den opregnede indtastning sikrer, at der leveres gyldige indstillinger til pipelinen, såsom denne, der sendes til ConditionLessThanOrEqualTo trin længere ned:

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

For at oprette et databehandlingstrin, som her bruges til at udføre feature engineering, bruger vi SKLearnProcessor at indlæse og transformere datasættet. Vi passerer local_pipeline_session variabel til klassekonstruktøren, som instruerer workflow-trinnet til at køre i lokal tilstand:

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

Dernæst opretter vi vores første egentlige pipeline-trin, en ProcessingStep objekt, som importeret fra SageMaker SDK. Processorargumenterne returneres fra et kald til SKLearnProcessor run() metode. Dette workflow-trin kombineres med andre trin mod slutningen af ​​notebook'en for at angive rækkefølgen af ​​operationen inden for pipelinen.

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)

Dernæst leverer vi kode til at etablere et træningstrin ved først at instansiere en standardestimator ved hjælp af SageMaker SDK. Vi passerer det samme local_pipeline_session variabel til estimatoren, kaldet xgb_train, som sagemaker_session argument. Fordi vi ønsker at træne en XGBoost-model, skal vi generere en gyldig billed-URI ved at angive følgende parametre, inklusive rammen og flere versionsparametre:

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 valgfrit kalde yderligere estimatormetoder f.eks set_hyperparameters(), for at angive hyperparameterindstillinger for træningsjobbet. Nu hvor vi har konfigureret en estimator, er vi klar til at oprette det faktiske træningstrin. Endnu en gang importerer vi TrainingStep klasse fra SageMaker SDK-biblioteket:

from sagemaker.workflow.steps import TrainingStep

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

Dernæst bygger vi endnu et behandlingstrin til at udføre modelevaluering. Dette gøres ved at oprette en ScriptProcessor instans og bestået local_pipeline_session objekt som 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,
)

For at muliggøre implementering af den trænede model, enten til en SageMaker-endepunkt i realtid eller til en batchtransformation skal vi oprette en Model objekt ved at videregive modelartefakter, den korrekte billed-URI og eventuelt vores tilpassede slutningskode. Så passerer vi dette Model gøre indsigelse mod en ModelStep, som føjes til den lokale pipeline. Se følgende kode:

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

Dernæst opretter vi et batch-transformationstrin, hvor vi indsender et sæt funktionsvektorer og udfører inferens. Vi skal først skabe en Transformer objekt og bestå local_pipeline_session parameter til det. Så laver vi en TransformStep, sender de påkrævede argumenter, og føj dette til 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)

Til sidst vil vi tilføje en branchebetingelse til arbejdsgangen, så vi kun kører batchtransformation, hvis resultaterne af modelevalueringen opfylder vores kriterier. Vi kan angive dette betingede ved at tilføje en ConditionStep med en bestemt tilstandstype, f.eks ConditionLessThanOrEqualTo. Vi opregner derefter trinene for de to grene, og definerer i det væsentlige hvis/andet eller sandt/falsk-grenene af rørledningen. Hvis_trinene, der er angivet i ConditionStep (step_create_model, step_transform) køres, når betingelsen evalueres til 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ølgende diagram illustrerer denne betingede gren og de tilhørende if/else-trin. Der køres kun én gren baseret på resultatet af modelevalueringstrinnet sammenlignet i tilstandstrinnet.

Nu hvor vi har defineret alle vores trin, og de underliggende klasseforekomster er oprettet, kan vi kombinere dem til en pipeline. Vi angiver nogle parametre og definerer operationsrækkefølgen helt afgørende ved blot at angive trinene i den ønskede rækkefølge. Bemærk, at TransformStep vises ikke her, fordi det er målet for det betingede trin og blev leveret som trinargument til ConditionalStep tidligere.

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

For at køre pipelinen skal du kalde to metoder: pipeline.upsert(), som uploader pipelinen til den underliggende tjeneste, og pipeline.start(), som begynder at køre pipelinen. Du kan bruge forskellige andre metoder til at undersøge kørselsstatus, liste pipeline-trinene og mere. Fordi vi brugte pipeline-sessionen i lokal tilstand, køres disse trin alle lokalt på din processor. Celleoutput under startmetoden viser output fra pipeline:

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

Du bør se en meddelelse nederst i celleoutput, der ligner følgende:

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

Vend tilbage til administrerede ressourcer

Efter at vi har bekræftet, at pipelinen kører uden fejl, og vi er tilfredse med flowet og formen af ​​pipelinen, kan vi genskabe pipelinen, men med SageMaker administrerede ressourcer og køre den igen. Den eneste ændring, der kræves, er at bruge PipelineSession objekt i stedet for LocalPipelineSession:

fra sagemaker.workflow.pipeline_context importer LocalPipelineSession
from sagemaker.workflow.pipeline_context import PipelineSession

local_pipeline_session = LocalPipelineSession()
pipeline_session = PipelineSession()

Dette informerer tjenesten om at køre hvert trin, der refererer til dette sessionsobjekt på SageMaker-administrerede ressourcer. I betragtning af den lille ændring illustrerer vi kun de nødvendige kodeændringer i den følgende kodecelle, men den samme ændring skal implementeres på hver celle ved hjælp af local_pipeline_session objekt. Ændringerne er dog identiske på tværs af alle celler, fordi vi kun erstatter 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 at det lokale sessionsobjekt er blevet erstattet overalt, genskaber vi pipelinen og kører den med SageMaker-administrerede ressourcer:

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

Ryd op

Hvis du vil holde ryddeligt i Studio-miljøet, kan du bruge følgende metoder til at slette SageMaker-pipelinen og modellen. Den fulde kode kan findes i eksemplet notesbog.

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

Konklusion

Indtil for nylig kunne du bruge den lokale tilstandsfunktion i SageMaker Processing og SageMaker Training til at gentage dine behandlings- og træningsscripts lokalt, før du kører dem på alle data med SageMaker-administrerede ressourcer. Med den nye lokaltilstandsfunktion i SageMaker Pipelines kan ML-udøvere nu anvende den samme metode, når de itererer på deres ML-pipelines, og sætter de forskellige ML-arbejdsgange sammen. Når pipelinen er klar til produktion, kræver det kun et par linjers kodeændringer at køre den med SageMaker-administrerede ressourcer. Dette reducerer pipeline-løbstiden under udvikling, hvilket fører til hurtigere pipeline-udvikling med hurtigere udviklingscyklusser, samtidig med at omkostningerne til SageMaker-administrerede ressourcer reduceres.

Hvis du vil vide mere, kan du besøge Amazon SageMaker Pipelines or Brug SageMaker Pipelines til at køre dine jobs lokalt.

Om forfatterne

Paul Hargis har fokuseret sin indsats på maskinlæring hos flere virksomheder, herunder AWS, Amazon og Hortonworks. Han nyder at bygge teknologiske løsninger og lære folk at få mest muligt ud af det. Før sin rolle hos AWS var han ledende arkitekt for Amazon Exports and Expansions, og hjalp amazon.com med at forbedre oplevelsen for internationale shoppere. Paul kan lide at hjælpe kunder med at udvide deres maskinlæringsinitiativer for at løse problemer i den virkelige verden.

Niklas Palme er Solutions Architect hos AWS i Stockholm, Sverige, hvor han hjælper kunder i hele Norden med at få succes i skyen. Han er især passioneret omkring serverløse teknologier sammen med IoT og maskinlæring. Uden for arbejdet er Niklas en ivrig langrendsløber og snowboarder samt mesteræggekoger.

Kirit Thadaka er en ML Solutions Architect, der arbejder i SageMaker Service SA-teamet. Før Kirit kom til AWS, arbejdede Kirit i AI-startups i tidlige stadier efterfulgt af nogen tid med rådgivning i forskellige roller inden for AI-forskning, MLOps og teknisk ledelse.