Bevált gyakorlatok és tervezési minták gépi tanulási munkafolyamatok építéséhez az Amazon SageMaker Pipelines | Amazon webszolgáltatások

Amazon SageMaker csővezetékek egy teljesen felügyelt AWS szolgáltatás gépi tanulási (ML) munkafolyamatok felépítéséhez és összehangolásához. A SageMaker Pipelines lehetővé teszi az ML alkalmazásfejlesztők számára az ML munkafolyamat különböző lépéseinek összehangolását, beleértve az adatbetöltést, az adatátalakítást, a képzést, a hangolást és a telepítést. Használhatja a SageMaker Pipelines ML-feladatokat a SageMakerben és annak alkalmazásában integráció a nagyobb AWS ökoszisztémával olyan erőforrások használatát is lehetővé teszi, mint a AWS Lambda függvények, Amazon EMR munkahelyek, és még sok más. Ez lehetővé teszi, hogy testreszabott és reprodukálható folyamatot hozzon létre az ML munkafolyamatok speciális követelményeihez.

Ebben a bejegyzésben bemutatunk néhány bevált gyakorlatot a SageMaker Pipelines értékének maximalizálásához és a fejlesztési élmény zökkenőmentessé tételéhez. Megbeszélünk néhány gyakori tervezési forgatókönyvet és mintát a SageMaker Pipelines építése során, és példákat adunk ezek megoldására.

A SageMaker Pipelines legjobb gyakorlatai

Ebben a részben bemutatunk néhány bevált gyakorlatot, amelyek követhetők a SageMaker Pipelines munkafolyamatok tervezése során. Ezek elfogadása javíthatja a fejlesztési folyamatot és ésszerűsítheti a SageMaker Pipelines működési irányítását.

Használja a Pipeline Session alkalmazást a csővezeték lusta betöltéséhez

Pipeline Session lehetővé teszi a folyamat erőforrásainak lusta inicializálását (a jobok csak a folyamat futási idejében indulnak el). A PipelineSession kontextus örökli a SageMaker Session és kényelmes módszereket valósít meg más SageMaker entitásokkal és erőforrásokkal való interakcióhoz, mint például a betanítási feladatok, végpontok, bemeneti adatkészletek Amazon egyszerű tárolási szolgáltatás (Amazon S3), és így tovább. A SageMaker Pipelines definiálásakor használja PipelineSession a szokásos SageMaker Session alatt:

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

Futtassa a folyamatokat helyi módban a költséghatékony és gyors iterációk érdekében a fejlesztés során

Futtathatja a csővezeték helyi módban használatával LocalPipelineSession kontextus. Ebben a módban a folyamat és a jobok helyileg futnak a helyi gép erőforrásaival, a SageMaker által felügyelt erőforrások helyett. A helyi mód költséghatékony módot biztosít a folyamatkódon való iterációra az adatok kisebb részhalmazával. A folyamat helyi tesztelése után méretezhető a futtatásra a következővel PipelineSession kontextus.

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

Kezelje a SageMaker folyamatot verziókezeléssel

A műtermékek és folyamatdefiníciók verziószáma általános követelmény a fejlesztési életciklusban. A folyamat több verzióját is létrehozhatja, ha a folyamat objektumokat egyedi előtaggal vagy utótaggal nevezi el, a leggyakoribb az időbélyeg, amint az a következő kódban látható:

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

A SageMaker kísérletekkel való integrációval megszervezheti és nyomon követheti a SageMaker folyamatokat

A SageMaker Pipelines könnyen integrálható SageMaker kísérletek szervezésére és nyomkövető csővezeték fut. Ezt megadásával érik el PipelineExperimentConfig létrehozásának időpontjában a csővezeték objektum. Ezzel a konfigurációs objektummal megadhat egy kísérletnevet és egy próbanevet. A SageMaker-folyamat futtatásának részletei a megadott kísérlet és próba alatt kerülnek rendszerezésre. Ha nem ad meg kifejezetten kísérletnevet, akkor a kísérlet neveként egy folyamatnevet használunk. Hasonlóképpen, ha nem ad meg kifejezetten egy próbaverziónevet, a próbafolyamat vagy a futtatási csoport neve egy folyamatazonosítót használ. Lásd a következő kódot:

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

A SageMaker folyamatok biztonságos futtatása privát VPC-n belül

Az ML-munkaterhelések biztonsága érdekében a legjobb gyakorlat, ha a SageMaker Pipelines által szervezett feladatokat biztonságos hálózati konfigurációban telepíti egy privát VPC-n, privát alhálózatokon és biztonsági csoportokon belül. A biztonságos környezet használatának biztosításához és érvényesítéséhez a következőket hajthatja végre AWS Identity and Access Management (IAM) szabályzata a SageMaker végrehajtó szerep (ezt a szerepet vállalja a folyamat során a folyamat). Hozzáadhatja azt a házirendet is, amely a SageMaker Pipelines által szervezett feladatokat hálózati elkülönített módban futtatja.

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

Példát a folyamat megvalósítására ezekkel a biztonsági ellenőrzésekkel a következő helyen talál Munkák összehangolása, modellregisztráció és folyamatos telepítés az Amazon SageMakerrel biztonságos környezetben.

Kövesse nyomon a folyamatok futtatásának költségeit címkék segítségével

A SageMaker csővezetékek használata önmagában ingyenes; fizetnie kell a számítási és tárolási erőforrásokért, amelyeket az egyes folyamatlépések, például a feldolgozás, a betanítás és a kötegelt következtetés részeként állít elő. A folyamatonkénti költségek összesítéséhez használhatja címkék minden erőforrást létrehozó folyamatlépésben. Ezekre a címkékre ezután hivatkozni lehet a költségfeltáróban a folyamat teljes üzemeltetési költségének szűréséhez és összesítéséhez, amint az a következő példában látható:

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

A költségfeltáróból most megkaphatja a költséget a címke alapján szűrve:

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

Tervezési minták néhány gyakori forgatókönyvhöz

Ebben a részben a SageMaker Pipelines néhány gyakori használati esetére vonatkozó tervezési mintákat tárgyaljuk.

Futtasson egy könnyű Python-függvényt Lambda lépéssel

A Python függvények mindenütt jelen vannak az ML munkafolyamatokban; előfeldolgozásban, utófeldolgozásban, értékelésben stb. használják. A Lambda egy kiszolgáló nélküli számítási szolgáltatás, amely lehetővé teszi a kód futtatását kiszolgálók kiépítése vagy kezelése nélkül. A Lambda segítségével kódot futtathat a kívánt nyelven, amely magában foglalja a Pythont is. Ezzel egyéni Python-kódot futtathat a folyamat részeként. Lambda lépés lehetővé teszi a Lambda funkciók futtatását a SageMaker folyamat részeként. Kezdje a következő kóddal:

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

Hozza létre a Lambda függvényt a A SageMaker Python SDK Lambda segítője:

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

Hívja a Lambda lépést:

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

Adatok átadása a lépések között

A folyamat lépéseihez tartozó bemeneti adatok vagy egy elérhető adathely, vagy a folyamatban lévő előző lépések valamelyike ​​által generált adatok. Ezt az információt megadhatja a ProcessingInput paraméter. Nézzünk meg néhány forgatókönyvet a ProcessingInput használatára vonatkozóan.

1. forgatókönyv: Adja át egy lambda-lépés kimenetét (primitív adattípusok) egy feldolgozási lépésnek

A primitív adattípusok olyan skaláris adattípusokra vonatkoznak, mint a string, integer, Boolean és float.

A következő kódrészlet egy Lambda függvényt határoz meg, amely primitív adattípusú változók szótárát adja vissza. A Lambda-függvénykód a kulcs-érték párokból álló JSON-kódot adja vissza, ha a SageMaker folyamat Lambda lépéséből hívja meg.

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

A folyamat definíciójában ezután megadhat egy adott adattípusú SageMaker folyamatparamétereket, és beállíthatja a változót a Lambda függvény kimenetére:

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. forgatókönyv: Adja át egy lambda-lépés kimenetét (nem primitív adattípusok) egy feldolgozási lépésnek

A nem primitív adattípusok nem skaláris adattípusokra utalnak (pl. NamedTuple). Előfordulhat olyan helyzet, amikor egy Lambda-függvényből nem primitív adattípust kell visszaadnia. Ehhez a nem primitív adattípust karakterláncra kell konvertálnia:

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

Ezután ezt a karakterláncot használhatja bemenetként a folyamat egy következő lépéséhez. A megnevezett sor használatához a kódban használja a eval() a Python kifejezés elemzéséhez a karakterláncban:

# 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. forgatókönyv: Adja át egy lépés kimenetét egy tulajdonságfájlon

A feldolgozási lépés kimenetét is tárolhatja a tulajdonság JSON fájl downstream fogyasztásra a ConditionStep vagy egy másik ProcessingStep. Használhatja a JSONGet függvény lekérdezni a ingatlan fájl. Lásd a következő kódot:

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

Tegyük fel, hogy a tulajdonságfájl tartalma a következő volt:

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

Ebben az esetben lekérdezhető egy adott érték, és a következő lépésekben használható a JsonGet függvény segítségével:

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

Paraméterezzen egy változót a folyamat definíciójában

Gyakran kívánatos a változók paraméterezése, hogy azok futás közben is használhatók legyenek – például egy S3 URI létrehozásához. Paraméterezhet egy karakterláncot úgy, hogy az kiértékelésre kerüljön futás közben a következővel Join funkció. A következő kódrészlet bemutatja, hogyan kell meghatározni a változót a Join függvényt, és ezzel állítsa be a kimeneti helyet egy feldolgozási lépésben:

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

Futtasson párhuzamos kódot egy iterálható elemen

Egyes ML-munkafolyamatok párhuzamos for-ciklusokban futtatják a kódot egy statikus elemkészleten (an megismételhető). Ez lehet ugyanaz a kód, amely különböző adatokon fut, vagy egy másik kódrészlet, amelyet minden egyes elemhez le kell futtatni. Például, ha egy fájlban nagyon sok sor van, és fel akarja gyorsítani a feldolgozási időt, támaszkodhat az előbbi mintára. Ha különböző átalakításokat szeretne végrehajtani az adatok bizonyos alcsoportjain, előfordulhat, hogy az adatok minden alcsoportjához más kódrészletet kell futtatnia. A következő két forgatókönyv bemutatja, hogyan tervezhet SageMaker-folyamatokat erre a célra.

1. forgatókönyv: Valósítson meg egy feldolgozási logikát az adatok különböző részein

Egy feldolgozási feladatot több példányban is futtathat (beállítással instance_count 1-nél nagyobb értékre). Ez elosztja az Amazon S3 bemeneti adatait az összes feldolgozási példányban. Ezután egy parancsfájl (process.py) segítségével dolgozhat az adatok egy meghatározott részén a példányszám és az elemek listájának megfelelő eleme alapján. A process.py programozási logikája felírható úgy, hogy a feldolgozott elemek listájától függően egy másik modul vagy kódrészlet fut le. A következő példa egy ProcessingStepben használható processzort határoz meg:

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. forgatókönyv: Futtassa le a lépések sorozatát

Ha van egy lépéssorozat, amelyet párhuzamosan kell futtatni, minden sorozatot független SageMaker folyamatként határozhat meg. Ezeknek a SageMaker-folyamatoknak a futtatása ezután elindítható egy Lambda függvényből, amely egy LambdaStep a szülőfolyamatban. A következő kódrészlet azt a forgatókönyvet szemlélteti, amikor két különböző SageMaker-folyamat fut:

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

Következtetés

Ebben a bejegyzésben megvitattunk néhány bevált gyakorlatot a SageMaker csővezetékek hatékony használatához és karbantartásához. Biztosítottunk bizonyos mintákat is, amelyeket átvehet a SageMaker Pipelines munkafolyamatok tervezése során, akár új folyamatokat készít, akár ML-munkafolyamatokat költöztet át más hangszerelési eszközökből. A SageMaker Pipelines ML-munkafolyamatok összehangolásához való használatának megkezdéséhez tekintse meg a kódminták a GitHubon és a Amazon SageMaker modellépítő csővezetékek.

A szerzőkről

Pinak Panigrahi együttműködik az ügyfelekkel, hogy gépi tanulásra támaszkodó megoldásokat építsenek a stratégiai üzleti problémák megoldására az AWS-en. Ha nem a gépi tanulással van elfoglalva, akkor kirándulni, könyvet olvasni vagy sportokat nézni.

Meenakshisundaram Thandavarayan az AWS-nél dolgozik AI/ML specialistaként. Szenvedélye az emberközpontú adat- és elemzési tapasztalatok tervezése, létrehozása és népszerűsítése. A Meena a fenntartható rendszerek fejlesztésére összpontosít, amelyek mérhető, versenyelőnyöket biztosítanak az AWS stratégiai ügyfelei számára. Meena egy összekötő, tervező, és arra törekszik, hogy az innováció, az inkubáció és a demokratizálódás révén új munkamódszerek felé terelje az üzletet.

• SEO által támogatott tartalom és PR terjesztés. Erősödjön még ma.
• PlatoData.Network Vertical Generative Ai. Erősítse meg magát. Hozzáférés itt.
• PlatoAiStream. Web3 Intelligence. Felerősített tudás. Hozzáférés itt.
• PlatoESG. Autóipar / elektromos járművek, Carbon, CleanTech, Energia, Környezet, Nap, Hulladékgazdálkodás. Hozzáférés itt.
• PlatoHealth. Biotechnológiai és klinikai vizsgálatok intelligencia. Hozzáférés itt.
• ChartPrime. Emelje fel kereskedési játékát a ChartPrime segítségével. Hozzáférés itt.
• BlockOffsets. A környezetvédelmi ellentételezési tulajdon korszerűsítése. Hozzáférés itt.
• Forrás: https://aws.amazon.com/blogs/machine-learning/best-practices-and-design-patterns-for-building-machine-learning-workflows-with-amazon-sagemaker-pipelines/