Praktik terbaik dan pola desain untuk membangun alur kerja pembelajaran mesin dengan Amazon SageMaker Pipelines | Layanan Web Amazon

Pipa Amazon SageMaker adalah layanan AWS yang terkelola sepenuhnya untuk membangun dan mengatur alur kerja pembelajaran mesin (ML). SageMaker Pipelines menawarkan kepada pengembang aplikasi ML kemampuan untuk mengatur berbagai langkah alur kerja ML, termasuk pemuatan data, transformasi data, pelatihan, penyetelan, dan penerapan. Anda dapat menggunakan SageMaker Pipelines untuk mengatur pekerjaan ML di SageMaker, dan itu integrasi dengan ekosistem AWS yang lebih besar juga memungkinkan Anda untuk menggunakan sumber daya seperti AWS Lambda fungsi, Amazon ESDM pekerjaan, dan banyak lagi. Hal ini memungkinkan Anda membangun pipeline yang disesuaikan dan dapat direproduksi untuk kebutuhan spesifik dalam alur kerja ML Anda.

Dalam postingan ini, kami memberikan beberapa praktik terbaik untuk memaksimalkan nilai SageMaker Pipelines dan membuat pengalaman pengembangan menjadi lancar. Kami juga membahas beberapa skenario dan pola desain umum saat membangun SageMaker Pipelines dan memberikan contoh untuk mengatasinya.

Praktik terbaik untuk SageMaker Pipelines

Di bagian ini, kami membahas beberapa praktik terbaik yang dapat diikuti saat merancang alur kerja menggunakan SageMaker Pipelines. Mengadopsinya dapat meningkatkan proses pengembangan dan menyederhanakan manajemen operasional SageMaker Pipelines.

Gunakan Sesi Pipeline untuk pemuatan pipeline yang lambat

Sesi Saluran memungkinkan inisialisasi sumber daya pipa yang lambat (pekerjaan tidak dimulai sampai runtime pipa). Itu PipelineSession konteks mewarisi Sesi SageMaker dan menerapkan metode yang mudah digunakan untuk berinteraksi dengan entitas dan sumber daya SageMaker lainnya, seperti tugas pelatihan, titik akhir, masukan kumpulan data di Layanan Penyimpanan Sederhana Amazon (Amazon S3), dan seterusnya. Saat mendefinisikan SageMaker Pipelines, Anda harus menggunakan PipelineSession selama Sesi SageMaker reguler:

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

Jalankan pipeline dalam mode lokal untuk iterasi yang hemat biaya dan cepat selama pengembangan

Anda dapat menjalankan pipa dalam mode lokal menggunakan LocalPipelineSession konteks. Dalam mode ini, alur dan pekerjaan dijalankan secara lokal menggunakan sumber daya di mesin lokal, bukan sumber daya yang dikelola SageMaker. Mode lokal menyediakan cara hemat biaya untuk melakukan iterasi pada kode pipeline dengan subkumpulan data yang lebih kecil. Setelah pipeline diuji secara lokal, pipeline dapat diskalakan untuk dijalankan menggunakan Sesi Pipa konteks.

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

Kelola alur SageMaker melalui pembuatan versi

Pembuatan versi artefak dan definisi alur merupakan persyaratan umum dalam siklus hidup pengembangan. Anda dapat membuat beberapa versi alur dengan memberi nama objek alur dengan awalan atau akhiran unik, yang paling umum adalah stempel waktu, seperti yang ditunjukkan dalam kode berikut:

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

Atur dan lacak alur SageMaker yang berjalan dengan mengintegrasikan dengan Eksperimen SageMaker

SageMaker Pipelines dapat diintegrasikan dengan mudah Eksperimen SageMaker untuk pengorganisasian dan pelacakan jalur pipa berjalan. Hal ini dicapai dengan menentukan Konfigurasi Eksperimen Saluran Pipa pada saat pembuatan a objek pipa. Dengan objek konfigurasi ini, Anda dapat menentukan nama eksperimen dan nama uji coba. Detail proses alur SageMaker diatur berdasarkan eksperimen dan uji coba yang ditentukan. Jika Anda tidak menentukan nama eksperimen secara eksplisit, nama alur akan digunakan untuk nama eksperimen. Demikian pula, jika Anda tidak secara eksplisit menentukan nama uji coba, ID eksekusi alur akan digunakan untuk nama grup uji coba atau eksekusi. Lihat kode berikut:

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

Jalankan pipeline SageMaker dengan aman dalam VPC pribadi

Untuk mengamankan beban kerja ML, praktik terbaiknya adalah menerapkan pekerjaan yang diatur oleh SageMaker Pipelines dalam konfigurasi jaringan yang aman dalam VPC privat, subnet privat, dan grup keamanan. Untuk memastikan dan menegakkan penggunaan lingkungan aman ini, Anda dapat menerapkan hal berikut Identitas AWS dan Manajemen Akses (IAM) kebijakan untuk Peran eksekusi SageMaker (ini adalah peran yang diambil oleh pipeline selama dijalankan). Anda juga dapat menambahkan kebijakan untuk menjalankan pekerjaan yang diatur oleh SageMaker Pipelines dalam mode isolasi jaringan.

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

Untuk contoh penerapan jalur pipa dengan kontrol keamanan ini, lihat Mengatur Pekerjaan, Pendaftaran Model, dan Penerapan Berkelanjutan dengan Amazon SageMaker di lingkungan yang aman.

Pantau biaya pengoperasian alur menggunakan tag

Menggunakan saluran pipa SageMaker sendiri tidak dipungut biaya; Anda membayar sumber daya komputasi dan penyimpanan yang Anda jalankan sebagai bagian dari langkah-langkah pipeline individual seperti pemrosesan, pelatihan, dan inferensi batch. Untuk menggabungkan biaya per pengoperasian alur, Anda dapat menyertakan tag di setiap langkah alur yang menciptakan sumber daya. Tag ini kemudian dapat direferensikan di penjelajah biaya untuk memfilter dan menggabungkan total biaya pengoperasian alur, seperti yang ditunjukkan dalam contoh berikut:

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

Dari penjelajah biaya, kini Anda bisa mendapatkan biaya yang difilter berdasarkan tag:

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

Pola desain untuk beberapa skenario umum

Di bagian ini, kami membahas pola desain untuk beberapa kasus penggunaan umum dengan SageMaker Pipelines.

Jalankan fungsi Python ringan menggunakan langkah Lambda

Fungsi Python ada di mana-mana dalam alur kerja ML; mereka digunakan dalam prapemrosesan, pascapemrosesan, evaluasi, dan banyak lagi. Lambda adalah layanan komputasi tanpa server yang memungkinkan Anda menjalankan kode tanpa menyediakan atau mengelola server. Dengan Lambda, Anda dapat menjalankan kode dalam bahasa pilihan Anda yang mencakup Python. Anda dapat menggunakan ini untuk menjalankan kode Python khusus sebagai bagian dari saluran Anda. Sebuah langkah Lambda memungkinkan Anda menjalankan fungsi Lambda sebagai bagian dari saluran SageMaker Anda. Mulailah dengan kode berikut:

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

Buat fungsi Lambda menggunakan Pembantu Lambda 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")

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

Melewati data antar langkah

Data masukan untuk langkah alur dapat berupa lokasi data yang dapat diakses atau data yang dihasilkan oleh salah satu langkah sebelumnya dalam alur. Anda dapat memberikan informasi ini sebagai a ProcessingInput parameter. Mari kita lihat beberapa skenario bagaimana Anda dapat menggunakan ProcessingInput.

Skenario 1: Meneruskan output (tipe data primitif) dari langkah Lambda ke langkah pemrosesan

Tipe data primitif mengacu pada tipe data skalar seperti string, integer, Boolean, dan float.

Cuplikan kode berikut mendefinisikan fungsi Lambda yang mengembalikan kamus variabel dengan tipe data primitif. Kode fungsi Lambda Anda akan mengembalikan JSON pasangan nilai kunci ketika dipanggil dari langkah Lambda dalam alur SageMaker.

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

Dalam definisi alur, Anda kemudian dapat menentukan parameter alur SageMaker yang merupakan tipe data tertentu dan mengatur variabel ke output fungsi 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,
)

Skenario 2: Meneruskan output (tipe data non-primitif) dari langkah Lambda ke langkah pemrosesan

Tipe data non-primitif mengacu pada tipe data non-skalar (misalnya, NamedTuple). Anda mungkin memiliki skenario ketika Anda harus mengembalikan tipe data non-primitif dari fungsi Lambda. Untuk melakukan ini, Anda harus mengonversi tipe data non-primitif menjadi string:

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

Kemudian Anda dapat menggunakan string ini sebagai masukan untuk langkah berikutnya dalam alur. Untuk menggunakan tupel bernama dalam kode, gunakan eval() untuk mengurai ekspresi Python dalam string:

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

Skenario 3: Meneruskan keluaran suatu langkah melalui file properti

Anda juga dapat menyimpan output dari langkah pemrosesan di a file JSON properti untuk konsumsi hilir di a ConditionStep atau lainnya ProcessingStep. Anda dapat menggunakan Fungsi JSONGet untuk menanyakan a berkas properti. Lihat kode berikut:

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

Anggaplah isi file properti adalah sebagai berikut:

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

Dalam hal ini, nilai tertentu dapat ditanyakan dan digunakan pada langkah selanjutnya menggunakan fungsi JsonGet:

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

Parameterisasikan variabel dalam definisi alur

Membuat parameter variabel agar dapat digunakan saat runtime sering kali diinginkan—misalnya, untuk membuat URI S3. Anda dapat membuat parameter string sedemikian rupa sehingga dievaluasi saat runtime menggunakan Join fungsi. Cuplikan kode berikut menunjukkan cara mendefinisikan variabel menggunakan Join fungsi dan gunakan itu untuk mengatur lokasi keluaran dalam langkah pemrosesan:

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

Jalankan kode paralel melalui iterable

Beberapa alur kerja ML menjalankan kode secara paralel untuk loop pada sekumpulan item statis (an bisa diubah). Ini bisa berupa kode yang sama yang dijalankan pada data berbeda atau bagian kode berbeda yang perlu dijalankan untuk setiap item. Misalnya, jika Anda memiliki jumlah baris yang sangat banyak dalam sebuah file dan ingin mempercepat waktu pemrosesan, Anda dapat mengandalkan pola sebelumnya. Jika Anda ingin melakukan transformasi berbeda pada subgrup tertentu dalam data, Anda mungkin harus menjalankan potongan kode berbeda untuk setiap subgrup dalam data. Dua skenario berikut menggambarkan bagaimana Anda dapat merancang alur SageMaker untuk tujuan ini.

Skenario 1: Menerapkan logika pemrosesan pada bagian data yang berbeda

Anda dapat menjalankan pekerjaan pemrosesan dengan beberapa contoh (dengan pengaturan instance_count ke nilai lebih besar dari 1). Ini mendistribusikan data masukan dari Amazon S3 ke semua instans pemrosesan. Anda kemudian dapat menggunakan skrip (process.py) untuk mengerjakan bagian data tertentu berdasarkan nomor instans dan elemen terkait dalam daftar item. Logika pemrograman di process.py dapat ditulis sedemikian rupa sehingga modul atau potongan kode yang berbeda dijalankan bergantung pada daftar item yang diprosesnya. Contoh berikut mendefinisikan prosesor yang dapat digunakan dalam 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" ) ], ]
)

Skenario 2: Jalankan serangkaian langkah

Saat Anda memiliki rangkaian langkah yang perlu dijalankan secara paralel, Anda dapat menentukan setiap rangkaian sebagai alur SageMaker independen. Jalannya alur SageMaker ini kemudian dapat dipicu dari fungsi Lambda yang merupakan bagian dari a LambdaStep di pipa induk. Potongan kode berikut mengilustrasikan skenario di mana dua alur SageMaker yang berbeda dipicu:

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

Kesimpulan

Dalam postingan ini, kami membahas beberapa praktik terbaik untuk penggunaan dan pemeliharaan pipeline SageMaker secara efisien. Kami juga menyediakan pola tertentu yang dapat Anda terapkan saat merancang alur kerja dengan SageMaker Pipelines, baik Anda membuat alur baru atau memigrasikan alur kerja ML dari alat orkestrasi lainnya. Untuk memulai SageMaker Pipelines untuk orkestrasi alur kerja ML, lihat contoh kode di GitHub dan Pipa Pembuatan Model Amazon SageMaker.

Tentang Penulis

Pinak Panigrahi bekerja dengan pelanggan untuk membangun solusi berbasis pembelajaran mesin untuk memecahkan masalah bisnis strategis di AWS. Saat tidak sibuk dengan pembelajaran mesin, ia dapat ditemukan sedang berjalan-jalan, membaca buku, atau menonton olahraga.

Meenakshisundaram Thandavarayan bekerja untuk AWS sebagai Spesialis AI/ ML. Dia memiliki hasrat untuk merancang, membuat, dan mempromosikan pengalaman data dan analitik yang berpusat pada manusia. Meena berfokus pada pengembangan sistem berkelanjutan yang memberikan keunggulan kompetitif yang terukur bagi pelanggan strategis AWS. Meena adalah penghubung, pemikir desain, dan berusaha untuk mengarahkan bisnis ke cara kerja baru melalui inovasi, inkubasi, dan demokratisasi.

• Konten Bertenaga SEO & Distribusi PR. Dapatkan Amplifikasi Hari Ini.
• PlatoData.Jaringan Vertikal Generatif Ai. Berdayakan Diri Anda. Akses Di Sini.
• PlatoAiStream. Intelijen Web3. Pengetahuan Diperkuat. Akses Di Sini.
• PlatoESG. Otomotif / EV, Karbon, teknologi bersih, energi, Lingkungan Hidup, Tenaga surya, Penanganan limbah. Akses Di Sini.
• PlatoHealth. Kecerdasan Uji Coba Biotek dan Klinis. Akses Di Sini.
• ChartPrime. Tingkatkan Game Trading Anda dengan ChartPrime. Akses Di Sini.
• BlockOffset. Modernisasi Kepemilikan Offset Lingkungan. Akses Di Sini.
• Sumber: https://aws.amazon.com/blogs/machine-learning/best-practices-and-design-patterns-for-building-machine-learning-workflows-with-amazon-sagemaker-pipelines/