Parimad tavad ja disainimustrid masinõppe töövoogude loomiseks Amazon SageMaker Pipelines | Amazoni veebiteenused

Amazon SageMakeri torujuhtmed on täielikult hallatav AWS-teenus masinõppe (ML) töövoogude ehitamiseks ja korraldamiseks. SageMaker Pipelines pakub ML-i rakenduste arendajatele võimalust korraldada ML-i töövoo erinevaid etappe, sealhulgas andmete laadimist, andmete teisendamist, koolitust, häälestamist ja juurutamist. Saate kasutada SageMakeri torujuhtmeid ML-tööde korraldamiseks SageMakeris ja selles integratsioon suurema AWS-ökosüsteemiga võimaldab kasutada ka selliseid ressursse nagu AWS Lambda funktsioonid Amazon EMR töökohti ja palju muud. See võimaldab teil luua kohandatud ja reprodutseeritava torujuhtme, mis vastab teie ML-i töövoogude konkreetsetele nõuetele.

Selles postituses pakume välja mõned parimad tavad SageMakeri torujuhtmete väärtuse maksimeerimiseks ja arenduskogemuse sujuvaks muutmiseks. Samuti käsitleme SageMakeri torujuhtmete ehitamisel mõningaid levinumaid disainistsenaariume ja mustreid ning pakume näiteid nende lahendamiseks.

SageMakeri torujuhtmete parimad tavad

Selles jaotises käsitleme mõningaid parimaid tavasid, mida saab SageMakeri torujuhtmete abil töövoogude kavandamisel järgida. Nende vastuvõtmine võib parandada arendusprotsessi ja tõhustada SageMakeri torujuhtmete operatiivjuhtimist.

Torujuhtme laisaks laadimiseks kasutage Pipeline Sessioni

Torujuhtme seanss võimaldab konveieri ressursside laisalt initsialiseerida (töid ei käivitata enne konveieri tööaega). The PipelineSession kontekst pärib SageMakeri seanss ja rakendab mugavaid meetodeid teiste SageMakeri üksuste ja ressurssidega suhtlemiseks, nagu koolitustööd, lõpp-punktid, sisendandmed Amazoni lihtne salvestusteenus (Amazon S3) ja nii edasi. SageMakeri torujuhtmete määratlemisel peaksite kasutama PipelineSession tavalise SageMakeri seansi ajal:

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

Kulusäästlikuks ja kiireks iteratsiooniks arenduse ajal käitage torujuhtmeid kohalikus režiimis

Võite joosta a torujuhe kohalikus režiimis kasutades LocalPipelineSession Sisu. Selles režiimis käitatakse konveierit ja töid kohapeal, kasutades kohaliku masina ressursse, mitte SageMakeri hallatavaid ressursse. Kohalik režiim pakub kulutõhusat viisi konveieri koodi itereerimiseks väiksema andmehulgaga. Pärast seda, kui torujuhet on kohapeal testitud, saab seda skaleerida töötama, kasutades PipelineSession Sisu.

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

Hallake SageMakeri konveieri versioonimise kaudu

Artefaktide ja torujuhtme definitsioonide versioonide koostamine on arenduse elutsükli tavaline nõue. Saate luua torujuhtme mitu versiooni, nimetades konveieriobjektidele kordumatu ees- või järelliide, millest kõige tavalisem on ajatempel, nagu on näidatud järgmises koodis:

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

Korraldage ja jälgige SageMakeri konveieri käitamist, integreerides SageMaker Experiments

SageMakeri torujuhtmeid saab hõlpsasti integreerida SageMakeri katsed korraldamise eest ja torujuhtme kulgemise jälgimine. See saavutatakse täpsustades PipelineExperimentConfig loomise ajal a torujuhtme objekt. Selle konfiguratsiooniobjektiga saate määrata katse nime ja prooviversiooni nime. SageMakeri torujuhtme käitamise üksikasjad korraldatakse määratud katse ja prooviversiooni alusel. Kui te ei määra selgesõnaliselt katse nime, kasutatakse katse nimena torujuhtme nime. Samamoodi, kui te ei määra selgesõnaliselt prooviversiooni nime, kasutatakse prooviversiooni või käitamisrühma nime jaoks konveieri käitamise ID-d. Vaadake järgmist koodi:

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

Käitage SageMakeri torujuhtmeid turvaliselt privaatses VPC-s

ML-i töökoormuste turvamiseks on parim tava juurutada SageMaker Pipelinesi korraldatud tööd turvalises võrgukonfiguratsioonis privaatses VPC-s, privaatsetes alamvõrkudes ja turvarühmades. Selle turvalise keskkonna kasutamise tagamiseks ja jõustamiseks saate rakendada järgmist AWS-i identiteedi- ja juurdepääsuhaldus (IAM) poliitika jaoks SageMakeri täitmise roll (selle rolli võtab torujuhe selle käitamise ajal). Samuti saate lisada poliitika SageMaker Pipelinesi korraldatud tööde käitamiseks võrgu eraldamise režiimis.

# 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" } } } ]
}

Näide torujuhtme rakendamisest, kui need turvakontrollid on paigas, leiate aadressilt Tööde korraldamine, mudeli registreerimine ja pidev juurutamine Amazon SageMakeriga turvalises keskkonnas.

Jälgige konveieri käitamise kulusid siltide abil

SageMakeri torujuhtmete kasutamine iseenesest on tasuta; maksate arvutus- ja salvestusressursside eest, mille loote üksikute konveieri etappide (nt töötlemine, koolitus ja partii järeldused) raames. Kulude koondamiseks torujuhtme käitamise kohta saate lisada silte igas konveieri etapis, mis loob ressurssi. Nendele siltidele saab seejärel kuluuurijas viidata, et filtreerida ja koondada torujuhtme käitamise kogukulu, nagu on näidatud järgmises näites.

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

Kuluuurijast saate nüüd kulud sildi järgi filtreerida:

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', ] } }
)

Kujundage mustrid mõne levinud stsenaariumi jaoks

Selles jaotises käsitleme SageMakeri torujuhtmete mõningate levinud kasutusjuhtude disainimustreid.

Käivitage kerge Pythoni funktsioon, kasutades lambda sammu

Pythoni funktsioonid on ML-i töövoogudes kõikjal olemas; neid kasutatakse eeltöötlusel, järeltöötlusel, hindamisel jm. Lambda on serverita arvutusteenus, mis võimaldab teil koodi käitada ilma servereid pakkumata või haldamata. Lambda abil saate käivitada koodi eelistatud keeles, mis sisaldab Pythonit. Saate seda kasutada kohandatud Pythoni koodi käivitamiseks torujuhtme osana. Lambda samm võimaldab teil käitada Lambda funktsioone osana oma SageMakeri torustikust. Alustage järgmise koodiga:

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

Looge lambda funktsioon, kasutades SageMaker Python SDK Lambda abimees:

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

Helistage lambda sammule:

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

Andmete edastamine etappide vahel

Konveieri etapi sisendandmed on kas juurdepääsetav andmeasukoht või andmed, mis on genereeritud konveieri mõne eelneva etapiga. Saate selle teabe esitada kui a ProcessingInput parameeter. Vaatame mõnda stsenaariumi ProcessingInputi kasutamise kohta.

1. stsenaarium: edastage lambda-etapi väljund (primitiivsed andmetüübid) töötlemisetappi

Primitiivsed andmetüübid viitavad skalaarsetele andmetüüpidele, nagu string, täisarv, tõeväärtus ja ujuk.

Järgmine koodilõik määratleb Lambda funktsiooni, mis tagastab primitiivsete andmetüüpidega muutujate sõnastiku. Teie Lambda-funktsiooni kood tagastab võtme-väärtuste paaride JSON-i, kui see käivitatakse SageMakeri konveieri Lambda-etapi kaudu.

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

Seejärel saate konveieri määratluses määratleda SageMakeri konveieri parameetrid, mis on kindlat tüüpi andmetüüpi, ja määrata muutuja funktsiooni Lambda väljundiks:

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

2. stsenaarium: edastage lambda-etapi väljund (mitteprimitiivsed andmetüübid) töötlemisetappi

Mitteprimitiivsed andmetüübid viitavad mitteskalaarsetele andmetüüpidele (näiteks NamedTuple). Teil võib tekkida stsenaarium, kui peate Lambda-funktsioonist tagastama mitteprimitiivse andmetüübi. Selleks peate teisendama oma mitteprimitiivse andmetüübi stringiks:

# 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"]

Seejärel saate seda stringi kasutada konveieri järgmise etapi sisendina. Nimega korteeži kasutamiseks koodis kasutage eval() Pythoni avaldise sõelumiseks stringis:

# 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. stsenaarium: suunake sammu väljund läbi atribuudifaili

Saate salvestada ka töötlemisetapi väljundi a atribuudi JSON-fail alltarbimiseks a ConditionStep vm ProcessingStep. Võite kasutada Funktsioon JSONGet küsima a vara fail. Vaadake järgmist koodi:

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

Oletame, et atribuudifaili sisu oli järgmine:

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

Sel juhul saab seda küsida konkreetse väärtuse jaoks ja kasutada järgmistes etappides, kasutades funktsiooni JsonGet:

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

Konveieri määratluses muutuja parameetriks muutmine

Muutujate parameetrite muutmine nii, et neid saaks käitamise ajal kasutada, on sageli soovitav, näiteks S3 URI koostamiseks. Saate stringi parameetriks muuta nii, et seda hinnatakse käitusajal, kasutades Join funktsiooni. Järgmine koodilõik näitab, kuidas defineerida muutujat kasutades Join funktsiooni ja kasutage seda väljundi asukoha määramiseks töötlemisetapis:

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

Käivitage paralleelkood üle itereeritava

Mõned ML-i töövood käitavad koodi paralleelselt for-tsüklitena üle staatilise üksuste komplekti (an korratav). See võib olla sama kood, mida käitatakse erinevatel andmetel, või erinev koodilõik, mida tuleb iga üksuse jaoks käitada. Näiteks kui teil on failis väga palju ridu ja soovite töötlusaega kiirendada, võite tugineda endisele mustrile. Kui soovite teha andmete konkreetsetes alamrühmades erinevaid teisendusi, peate võib-olla käivitama iga andmete alamrühma jaoks erineva koodi. Järgmised kaks stsenaariumi illustreerivad, kuidas saate SageMakeri torujuhtmeid selleks otstarbeks kujundada.

1. stsenaarium: rakendage töötlemisloogikat andmete erinevatel osadel

Saate käitada töötlemistööd mitme eksemplariga (määrates instance_count väärtusele, mis on suurem kui 1). See jagab Amazon S3 sisendandmed kõikidesse töötlemisjuhtumitesse. Seejärel saate kasutada skripti (process.py), et töötada teatud andmete osaga eksemplari numbri ja üksuste loendi vastava elemendi alusel. Programmeerimisloogika failis process.py saab kirjutada nii, et olenevalt töödeldavate üksuste loendist käivitatakse erinev moodul või koodijupp. Järgmine näide määratleb protsessori, mida saab protsessis ProcessingStep kasutada:

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. stsenaarium: käivitage toimingute jada

Kui teil on toimingute jada, mida tuleb paralleelselt käivitada, saate iga jada määratleda iseseisva SageMakeri torujuhtmena. Nende SageMakeri torujuhtmete käitamise saab seejärel käivitada lambda funktsioonist, mis on osa LambdaStep emakonveieris. Järgmine koodilõik illustreerib stsenaariumi, kus käivitatakse kaks erinevat SageMakeri torujuhtme käitamist:

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

Järeldus

Selles postituses arutasime mõningaid parimaid tavasid SageMakeri torujuhtmete tõhusaks kasutamiseks ja hooldamiseks. Samuti pakkusime välja teatud mustrid, mida saate SageMaker Pipelinesiga töövoogude kujundamisel kasutada, olenemata sellest, kas loote uusi torujuhtmeid või migreerite ML-töövooge muudest orkestreerimistööriistadest. ML-i töövoo orkestreerimiseks SageMaker Pipelinesiga alustamiseks vaadake koodinäidised GitHubis ja Amazon SageMakeri mudeliehitustorud.

Autoritest

Pinak Panigrahi töötab koos klientidega, et luua masinõppepõhiseid lahendusi, et lahendada strateegilisi äriprobleeme AWS-is. Kui ta pole masinõppega hõivatud, võib teda leida matkamas, raamatut lugemas või sporti vaatamas.

Meenakshisundaram Thandavarayan töötab AWS-is AI/ML spetsialistina. Tal on kirg kujundada, luua ja edendada inimkeskseid andme- ja analüüsikogemusi. Meena keskendub jätkusuutlike süsteemide arendamisele, mis pakuvad AWS-i strateegilistele klientidele mõõdetavaid konkurentsieeliseid. Meena on ühendaja, disainimõtleja ja püüab innovatsiooni, inkubatsiooni ja demokratiseerimise kaudu juhtida äri uutele tööviisidele.

• SEO-põhise sisu ja PR-levi. Võimenduge juba täna.
• PlatoData.Network Vertikaalne generatiivne Ai. Jõustage ennast. Juurdepääs siia.
• PlatoAiStream. Web3 luure. Täiustatud teadmised. Juurdepääs siia.
• PlatoESG. Autod/elektrisõidukid, Süsinik, CleanTech, Energia, Keskkond päikeseenergia, Jäätmekäitluse. Juurdepääs siia.
• PlatoTervis. Biotehnoloogia ja kliiniliste uuringute luureandmed. Juurdepääs siia.
• ChartPrime. Tõsta oma kauplemismängu ChartPrime'iga kõrgemale. Juurdepääs siia.
• BlockOffsets. Keskkonnakompensatsiooni omandi ajakohastamine. Juurdepääs siia.
• Allikas: https://aws.amazon.com/blogs/machine-learning/best-practices-and-design-patterns-for-building-machine-learning-workflows-with-amazon-sagemaker-pipelines/