Najboljše prakse in oblikovalski vzorci za gradnjo delovnih tokov strojnega učenja z Amazon SageMaker Pipelines | Spletne storitve Amazon

Amazonski cevovodi SageMaker je v celoti upravljana storitev AWS za gradnjo in usmerjanje delovnih tokov strojnega učenja (ML). SageMaker Pipelines ponuja razvijalcem aplikacij ML možnost orkestriranja različnih korakov delovnega toka ML, vključno z nalaganjem podatkov, transformacijo podatkov, usposabljanjem, prilagajanjem in uvajanjem. SageMaker Pipelines lahko uporabite za orkestriranje opravil ML v SageMakerju in njegovem integracija z večjim ekosistemom AWS omogoča tudi uporabo virov, kot je AWS Lambda funkcije, Amazonski EMR delovna mesta in še več. To vam omogoča, da zgradite prilagojen in ponovljiv cevovod za posebne zahteve v vaših potekih dela ML.

V tej objavi ponujamo nekaj najboljših praks za povečanje vrednosti SageMaker Pipelines in omogočanje brezhibne razvojne izkušnje. Razpravljamo tudi o nekaterih pogostih načrtovalskih scenarijih in vzorcih pri izdelavi cevovodov SageMaker in nudimo primere za njihovo obravnavo.

Najboljše prakse za SageMaker Pipelines

V tem razdelku razpravljamo o nekaterih najboljših praksah, ki jih je mogoče upoštevati pri načrtovanju delovnih tokov z uporabo SageMaker Pipelines. Če jih sprejmete, lahko izboljšate razvojni proces in poenostavite operativno upravljanje SageMaker Pipelines.

Uporabite Pipeline Session za počasno nalaganje cevovoda

Pipeline Session omogoča leno inicializacijo virov cevovoda (opravila se ne zaženejo do izvajalnega časa cevovoda). The PipelineSession kontekst podeduje Seja SageMaker in izvaja priročne metode za interakcijo z drugimi entitetami in viri SageMaker, kot so delovna mesta za usposabljanje, končne točke, vhodni nizi podatkov v Preprosta storitev shranjevanja Amazon (Amazon S3) in tako naprej. Ko definirate cevovode SageMaker, uporabite PipelineSession na redni seji SageMaker:

from sagemaker.workflow.pipeline_context import PipelineSession
from sagemaker.sklearn.processing import SKLearnProcessor
role = sagemaker.get_execution_role()
pipeline_session = PipelineSession()
sklearn_processor = SKLearnProcessor( framework_version=’0.20.0’, instance_type=’ml.m5.xlarge’, instance_count=1, base_job_name="sklearn-abalone-process", role=role, sagemaker_session=pipeline_session,
)

Izvedite cevovode v lokalnem načinu za stroškovno učinkovite in hitre iteracije med razvojem

Lahko vodite a cevovod v lokalnem načinu uporabi LocalPipelineSession kontekstu. V tem načinu se cevovod in opravila izvajajo lokalno z uporabo virov na lokalnem računalniku namesto virov, ki jih upravlja SageMaker. Lokalni način zagotavlja stroškovno učinkovit način ponavljanja kode cevovoda z manjšo podmnožico podatkov. Ko je cevovod lokalno preizkušen, ga je mogoče prilagoditi za delovanje z uporabo PipelineSession kontekst.

from sagemaker.sklearn.processing import SKLearnProcessor
from sagemaker.workflow.pipeline_context import LocalPipelineSession
local_pipeline_session = LocalPipelineSession()
role = sagemaker.get_execution_role()
sklearn_processor = SKLearnProcessor( framework_version=’0.20.0’, instance_type=’ml.m5.xlarge, instance_count=1, base_job_name="sklearn-abalone-process", role=role, sagemaker_session=local_pipeline_session,
)

Upravljajte cevovod SageMaker prek različic

Različice artefaktov in definicij cevovoda so pogosta zahteva v življenjskem ciklu razvoja. Ustvarite lahko več različic cevovoda tako, da objekte cevovoda poimenujete z edinstveno predpono ali pripono, pri čemer je najpogostejši časovni žig, kot je prikazano v naslednji kodi:

from sagemaker.workflow.pipeline_context import PipelineSession
import time current_time = time.strftime("%Y-%m-%d-%H-%M-%S", time.gmtime())
pipeline_name = "pipeline_" + current_time
pipeline_session = PipelineSession()
pipeline = Pipeline( name=pipeline_name, steps=[step_process, step_train, step_eval, step_cond], sagemaker_session=pipeline_session,
)

Organizirajte in spremljajte potek cevovoda SageMaker z integracijo s SageMaker Experiments

SageMaker Pipelines je mogoče enostavno integrirati Poskusi SageMaker za organiziranje in sledenje potekom cevovoda. To se doseže z določitvijo PipelineExperimentConfig v času ustvarjanja a objekt cevovoda. S tem konfiguracijskim objektom lahko določite ime poskusa in ime poskusa. Podrobnosti o izvajanju cevovoda SageMaker so organizirane v skladu z določenim poskusom in preskusom. Če izrecno ne navedete imena preizkusa, se za ime preizkusa uporabi ime cevovoda. Podobno, če izrecno ne podate preskusnega imena, se za ime poskusne ali izvajalne skupine uporabi ID cevovoda. Oglejte si naslednjo kodo:

Pipeline( name="MyPipeline", parameters=[...], pipeline_experiment_config=PipelineExperimentConfig( experiment_name = ExecutionVariables.PIPELINE_NAME, trial_name = ExecutionVariables.PIPELINE_EXECUTION_ID ), steps=[...]
)

Varno zaženite cevovode SageMaker znotraj zasebnega VPC

Da bi zavarovali delovne obremenitve ML, je najboljša praksa razmestiti opravila, ki jih orkestrira SageMaker Pipelines, v varni konfiguraciji omrežja znotraj zasebnega VPC, zasebnih podomrežij in varnostnih skupin. Če želite zagotoviti in uveljaviti uporabo tega varnega okolja, lahko implementirate naslednje AWS upravljanje identitete in dostopa (IAM) politika za Izvršilna vloga SageMaker (to je vloga, ki jo prevzame cevovod med svojim delovanjem). Dodate lahko tudi pravilnik za izvajanje opravil, ki jih orkestrira SageMaker Pipelines, v načinu izolacije omrežja.

# IAM Policy to enforce execution within a private VPC { "Action": [ "sagemaker:CreateProcessingJob", "sagemaker:CreateTrainingJob", "sagemaker:CreateModel" ], "Resource": "*", "Effect": "Deny", "Condition": { "Null": { "sagemaker:VpcSubnets": "true" } }
} # IAM Policy to enforce execution in network isolation mode
{ "Version": "2012-10-17", "Statement": [ { "Effect": "Deny", "Action": [ "sagemaker:Create*" ], "Resource": "*", "Condition": { "StringNotEqualsIfExists": { "sagemaker:NetworkIsolation": "true" } } } ]
}

Za primer izvajanja cevovoda s temi varnostnimi kontrolami glejte Orkestriranje delovnih mest, registracije modela in neprekinjenega uvajanja z Amazon SageMaker v varnem okolju.

Spremljajte stroške cevovoda z uporabo oznak

Sama uporaba cevovodov SageMaker je brezplačna; plačate za računalniške in pomnilniške vire, ki jih zavrtite kot del posameznih korakov cevovoda, kot so obdelava, usposabljanje in paketno sklepanje. Če želite združiti stroške na potek cevovoda, lahko vključite oznake v vsakem koraku cevovoda, ki ustvari vir. Na te oznake se lahko nato sklicujete v raziskovalcu stroškov za filtriranje in združevanje skupnih stroškov delovanja cevovoda, kot je prikazano v naslednjem primeru:

sklearn_processor = SKLearnProcessor( framework_version=’0.20.0’, instance_type=’ml.m5.xlarge, instance_count=1, base_job_name="sklearn-abalone-process", role=role, tags=[{'Key':'pipeline-cost-tag', 'Value':'<<tag_parameter>>'}]
) step_process = ProcessingStep( name="AbaloneProcess", processor=sklearn_processor, ...
)

Iz raziskovalca stroškov lahko zdaj dobite stroške, filtrirane po oznaki:

response = client.get_cost_and_usage( TimePeriod={ 'Start': '2023-07-01', 'End': '2023-07-15' }, Metrics=['BLENDED_COST','USAGE_QUANTITY','UNBLENDED_COST'], Granularity='MONTHLY', Filter={ 'Dimensions': { 'Key':'USAGE_TYPE', 'Values': [ ‘SageMaker:Pipeline’ ] }, 'Tags': { 'Key': 'keyName', 'Values': [ 'keyValue', ] } }
)

Oblikovalski vzorci za nekatere običajne scenarije

V tem razdelku razpravljamo o načrtovalskih vzorcih za nekatere običajne primere uporabe cevovodov SageMaker.

Zaženite lahkotno funkcijo Python z uporabo koraka Lambda

Funkcije Pythona so vseprisotne v potekih dela ML; uporabljajo se pri predobdelavi, naknadni obdelavi, vrednotenju itd. Lambda je računalniška storitev brez strežnika, ki vam omogoča izvajanje kode brez zagotavljanja ali upravljanja strežnikov. Z Lambdo lahko izvajate kodo v svojem želenem jeziku, ki vključuje Python. To lahko uporabite za izvajanje kode Python po meri kot del vašega cevovoda. Lambda korak vam omogoča zagon funkcij Lambda kot del vašega cevovoda SageMaker. Začnite z naslednjo kodo:

%%writefile lambdafunc.py import json def lambda_handler(event, context): str1 = event["str1"] str2 = event["str2"] str3 = str1 + str2 return { "str3": str3 }

Ustvarite funkcijo Lambda z uporabo Lambda pomočnik SageMaker Python SDK:

from sagemaker.lambda_helper import Lambda def create_lambda(function_name, script, handler): response = Lambda( function_name=function_name, execution_role_arn=role, script= script, handler=handler, timeout=600, memory_size=10240, ).upsert() function_arn = response['FunctionArn'] return function_arn fn_arn = create_Lambda("func", "lambdafunc.py", handler = "lambdafunc.lambda_handler")

Pokličite korak Lambda:

from sagemaker.lambda_helper import Lambda
from sagemaker.workflow.lambda_step import ( LambdaStep, LambdaOutput, LambdaOutputTypeEnum
) str3 = LambdaOutput(output_name="str3", output_type=LambdaOutputTypeEnum.String) # Lambda Step
step_lambda1 = LambdaStep( name="LambdaStep1", lambda_func=Lambda( function_arn=fn_arn ), inputs={ "str1": "Hello", "str2": " World" }, outputs=[str3],
)

Prenos podatkov med koraki

Vhodni podatki za korak cevovoda so bodisi dostopna podatkovna lokacija ali podatki, ustvarjeni z enim od prejšnjih korakov v cevovodu. Te informacije lahko posredujete kot a ProcessingInput parameter. Oglejmo si nekaj scenarijev, kako lahko uporabite ProcessingInput.

Scenarij 1: Prenesite izhod (primitivni tipi podatkov) koraka Lambda v korak obdelave

Primitivni podatkovni tipi se nanašajo na skalarne podatkovne tipe, kot so niz, celo število, logični in plavajoči.

Naslednji delček kode definira funkcijo Lambda, ki vrne slovar spremenljivk s primitivnimi tipi podatkov. Vaša koda funkcije Lambda bo vrnila JSON parov ključ-vrednost, ko bo priklicana iz koraka Lambda v cevovodu SageMaker.

def handler(event, context): ... return { "output1": "string_value", "output2": 1, "output3": True, "output4": 2.0, }

V definiciji cevovoda lahko nato definirate parametre cevovoda SageMaker, ki so določene vrste podatkov, in nastavite spremenljivko na izhod funkcije Lambda:

from sagemaker.workflow.lambda_step import ( LambdaStep, LambdaOutput, LambdaOutputTypeEnum
)
from sagemaker.workflow.pipeline_context import PipelineSession
from sagemaker.sklearn.processing import SKLearnProcessor role = sagemaker.get_execution_role()
pipeline_session = PipelineSession() # 1. Define the output params of the Lambda Step str_outputParam = LambdaOutput(output_name="output1", output_type=LambdaOutputTypeEnum.String)
int_outputParam = LambdaOutput(output_name"output2", output_type=LambdaOutputTypeEnum.Integer)
bool_outputParam = LambdaOutput(output_name"output3", output_type=LambdaOutputTypeEnum.Boolean)
float_outputParam = LambdaOutput(output_name"output4", output_type=LambdaOutputTypeEnum.Float) # 2. Lambda step invoking the lambda function and returns the Output step_lambda = LambdaStep( name="MyLambdaStep", lambda_func=Lambda( function_arn="arn:aws:lambda:us-west-2:123456789012:function:sagemaker_test_lambda", session=PipelineSession(), ), inputs={"arg1": "foo", "arg2": "foo1"}, outputs=[ str_outputParam, int_outputParam, bool_outputParam, float_outputParam ],
) # 3. Extract the output of the Lambda str_outputParam = step_lambda.properties.Outputs["output1"] # 4. Use it in a subsequent step. For ex. Processing step sklearn_processor = SKLearnProcessor( framework_version="0.23-1", instance_type="ml.m5.xlarge", instance_count=1, sagemaker_session=pipeline_session, role=role
) processor_args = sklearn_processor.run( code="code/preprocess.py", #python script to run arguments=["--input-args", str_outputParam]
) step_process = ProcessingStep( name="processstep1", step_args=processor_args,
)

Scenarij 2: Prenesite izhod (neprimitivni tipi podatkov) koraka Lambda v korak obdelave

Neprimitivni podatkovni tipi se nanašajo na neskalarne podatkovne tipe (npr. NamedTuple). Morda imate scenarij, ko morate iz funkcije Lambda vrniti neprimitivni podatkovni tip. Če želite to narediti, morate svoj neprimitivni podatkovni tip pretvoriti v niz:

# Lambda function code returning a non primitive data type from collections import namedtuple def lambda_handler(event, context): Outputs = namedtuple("Outputs", "sample_output") named_tuple = Outputs( [ {'output1': 1, 'output2': 2}, {'output3': 'foo', 'output4': 'foo1'} ] )
return{ "named_tuple_string": str(named_tuple)
}

#Pipeline step that uses the Lambda output as a “Parameter Input” output_ref = step_lambda.properties.Outputs["named_tuple_string"]

Nato lahko ta niz uporabite kot vhod za naslednji korak v cevovodu. Za uporabo imenovane tuple v kodi uporabite eval() za razčlenitev izraza Python v nizu:

# Decipher the string in your processing logic code import argparse
from collections import namedtuple Outputs = namedtuple("Outputs", "sample_output") if __name__ == "__main__": parser = argparse.ArgumentParser() parser.add_argument("--named_tuple_string", type=str, required=True) args = parser.parse_args() #use eval to obtain the named tuple from the string named_tuple = eval(args.named_tuple_string)

3. scenarij: posredujte izhod koraka skozi datoteko lastnosti

Izhod koraka obdelave lahko shranite tudi v a lastninska datoteka JSON za nadaljnjo porabo v a ConditionStep ali drugi ProcessingStep. Lahko uporabite Funkcija JSONGet povprašati a lastninska datoteka. Glej naslednjo kodo:

# 1. Define a Processor with a ProcessingOutput
sklearn_processor = SKLearnProcessor( framework_version="0.23-1", instance_type="ml.m5.xlarge", instance_count=1, base_job_name="sklearn-abalone-preprocess", sagemaker_session=session, role=sagemaker.get_execution_role(),
) step_args = sklearn_processor.run( outputs=[ ProcessingOutput( output_name="hyperparam", source="/opt/ml/processing/evaluation" ), ], code="./local/preprocess.py", arguments=["--input-data", "s3://my-input"],
) # 2. Define a PropertyFile where the output_name matches that with the one used in the Processor

hyperparam_report = PropertyFile( name="AbaloneHyperparamReport", output_name="hyperparam", path="hyperparam.json",
)

Predpostavimo, da je vsebina datoteke lastnosti naslednja:

{ "hyperparam": { "eta": { "value": 0.6 } }
}

V tem primeru je mogoče poizvedovati za določeno vrednost in jo uporabiti v naslednjih korakih s funkcijo JsonGet:

# 3. Query the property file
eta = JsonGet( step_name=step_process.name, property_file=hyperparam_report, json_path="hyperparam.eta.value",
)

Parametrirajte spremenljivko v definiciji cevovoda

Parametriranje spremenljivk, tako da jih je mogoče uporabiti med izvajanjem, je pogosto zaželeno – na primer za izdelavo S3 URI. Niz lahko parametrirate tako, da se ovrednoti med izvajanjem z uporabo Join funkcijo. Naslednji delček kode prikazuje, kako definirati spremenljivko z uporabo Join funkcijo in jo uporabite za nastavitev izhodne lokacije v koraku obdelave:

# define the variable to store the s3 URI
s3_location = Join( on="/", values=[ "s3:/", ParameterString( name="MyBucket", default_value="" ), "training", ExecutionVariables.PIPELINE_EXECUTION_ID ]
) # define the processing step
sklearn_processor = SKLearnProcessor( framework_version="1.2-1", instance_type="ml.m5.xlarge", instance_count=processing_instance_count, base_job_name=f"{base_job_prefix}/sklearn-abalone-preprocess", sagemaker_session=pipeline_session, role=role,
) # use the s3uri as the output location in processing step
processor_run_args = sklearn_processor.run( outputs=[ ProcessingOutput( output_name="train", source="/opt/ml/processing/train", destination=s3_location, ), ], code="code/preprocess.py"
) step_process = ProcessingStep( name="PreprocessingJob”, step_args=processor_run_args,
)

Izvedite vzporedno kodo nad iterable

Nekateri poteki dela ML izvajajo kodo v vzporednih zankah for prek statičnega nabora elementov (an ponovljivo). To je lahko ista koda, ki se izvaja na različnih podatkih, ali drugačen del kode, ki ga je treba izvesti za vsak element. Na primer, če imate zelo veliko vrstic v datoteki in želite pospešiti čas obdelave, se lahko zanesete na prejšnji vzorec. Če želite izvesti različne transformacije na določenih podskupinah v podatkih, boste morda morali zagnati drugačen del kode za vsako podskupino v podatkih. Naslednja dva scenarija prikazujeta, kako lahko načrtujete cevovode SageMaker za ta namen.

Scenarij 1: Izvedite logiko obdelave na različnih delih podatkov

Posel obdelave lahko izvajate z več primerki (z nastavitvijo instance_count na vrednost večjo od 1). To porazdeli vhodne podatke iz Amazon S3 v vse primerke obdelave. Nato lahko uporabite skript (process.py) za delo na določenem delu podatkov na podlagi številke primerka in ustreznega elementa na seznamu elementov. Programsko logiko v procesu.py je mogoče napisati tako, da se zažene drug modul ali del kode, odvisno od seznama elementov, ki jih obdeluje. Naslednji primer definira procesor, ki ga je mogoče uporabiti v ProcessingStep:

sklearn_processor = FrameworkProcessor( estimator_cls=sagemaker.sklearn.estimator.SKLearn, framework_version="0.23-1", instance_type='ml.m5.4xlarge', instance_count=4, #number of parallel executions / instances base_job_name="parallel-step", sagemaker_session=session, role=role,
) step_args = sklearn_processor.run( code='process.py', arguments=[ "--items", list_of_items, #data structure containing a list of items inputs=[ ProcessingInput(source="s3://sagemaker-us-east-1-xxxxxxxxxxxx/abalone/abalone-dataset.csv", destination="/opt/ml/processing/input" ) ], ]
)

2. scenarij: Zaženite zaporedje korakov

Ko imate zaporedje korakov, ki jih je treba izvajati vzporedno, lahko vsako zaporedje definirate kot neodvisen cevovod SageMaker. Zagon teh cevovodov SageMaker se lahko nato sproži iz funkcije Lambda, ki je del LambdaStep v nadrejenem cevovodu. Naslednji del kode ponazarja scenarij, kjer se sprožita dva različna izvajanja cevovoda SageMaker:

import boto3
def lambda_handler(event, context): items = [1, 2] #sagemaker client sm_client = boto3.client("sagemaker") #name of the pipeline that needs to be triggered. #if there are multiple, you can fetch available pipelines using boto3 api #and trigger the appropriate one based on your logic. pipeline_name = 'child-pipeline-1' #trigger pipeline for every item response_ppl = sm_client.start_pipeline_execution( PipelineName=pipeline_name, PipelineExecutionDisplayName=pipeline_name+'-item-%d' %(s), ) pipeline_name = 'child-pipeline-2' response_ppl = sm_client.start_pipeline_execution( PipelineName=pipeline_name, PipelineExecutionDisplayName=pipeline_name+'-item-%d' %(s), )
return

zaključek

V tej objavi smo razpravljali o nekaterih najboljših praksah za učinkovito uporabo in vzdrževanje cevovodov SageMaker. Zagotovili smo tudi določene vzorce, ki jih lahko sprejmete med načrtovanjem delovnih tokov s SageMaker Pipelines, ne glede na to, ali ustvarjate nove cevovode ali selite delovne tokove ML iz drugih orodij za orkestracijo. Če želite začeti uporabljati SageMaker Pipelines za orkestracijo poteka dela ML, glejte vzorci kode na GitHubu in Cevovodi za gradnjo modelov Amazon SageMaker.

O avtorjih

Pinak Panigrahi sodeluje s strankami pri izdelavi rešitev, ki temeljijo na strojnem učenju, za reševanje strateških poslovnih problemov na AWS. Ko se ne ukvarja s strojnim učenjem, ga lahko najdemo na pohodu, bere knjigo ali gleda šport.

Meenakshisundaram Thandavarayan dela za AWS kot strokovnjak za AI/ML. Ima strast do oblikovanja, ustvarjanja in promocije na človeka osredotočenih podatkov in analitičnih izkušenj. Meena se osredotoča na razvoj trajnostnih sistemov, ki zagotavljajo merljive konkurenčne prednosti za strateške stranke AWS. Meena je povezovalka, oblikovalska misleca in si prizadeva usmeriti podjetja v nove načine dela z inovacijami, inkubacijo in demokratizacijo.

• Distribucija vsebine in PR s pomočjo SEO. Okrepite se še danes.
• PlatoData.Network Vertical Generative Ai. Opolnomočite se. Dostopite tukaj.
• PlatoAiStream. Web3 Intelligence. Razširjeno znanje. Dostopite tukaj.
• PlatoESG. Avtomobili/EV, Ogljik, CleanTech, Energija, Okolje, sončna energija, Ravnanje z odpadki. Dostopite tukaj.
• PlatoHealth. Obveščanje o biotehnologiji in kliničnih preskušanjih. Dostopite tukaj.
• ChartPrime. Izboljšajte svojo igro trgovanja s ChartPrime. Dostopite tukaj.
• BlockOffsets. Posodobitev okoljskega offset lastništva. Dostopite tukaj.
• vir: https://aws.amazon.com/blogs/machine-learning/best-practices-and-design-patterns-for-building-machine-learning-workflows-with-amazon-sagemaker-pipelines/