Найкращі практики та шаблони проектування для створення робочих процесів машинного навчання за допомогою Amazon SageMaker Pipelines | Веб-сервіси Amazon

Трубопроводи Amazon SageMaker це повністю керований сервіс AWS для створення та оркестрування робочих процесів машинного навчання (ML). SageMaker Pipelines пропонує розробникам додатків ML можливість організовувати різні етапи робочого процесу ML, включаючи завантаження даних, перетворення даних, навчання, налаштування та розгортання. Ви можете використовувати SageMaker Pipelines для оркестрування завдань машинного навчання в SageMaker і його інтеграція з більшою екосистемою AWS також дозволяє використовувати такі ресурси, як AWS Lambda функції, Amazon EMR робочі місця тощо. Це дає змогу створювати налаштований і відтворюваний конвеєр для конкретних вимог у ваших робочих процесах машинного навчання.

У цій публікації ми надаємо деякі найкращі практики, щоб максимізувати цінність SageMaker Pipelines і зробити процес розробки бездоганним. Ми також обговорюємо деякі загальні сценарії та шаблони проектування під час створення конвеєрів SageMaker і надаємо приклади їх вирішення.

Найкращі практики для SageMaker Pipelines

У цьому розділі ми обговорюємо деякі найкращі практики, яких можна дотримуватися під час проектування робочих процесів за допомогою конвеєрів SageMaker. Їх застосування може покращити процес розробки та оптимізувати операційне керування SageMaker Pipelines.

Використовуйте Pipeline Session для відкладеного завантаження конвеєра

Конвеєрна сесія дозволяє відкладену ініціалізацію ресурсів конвеєра (завдання не запускаються до моменту виконання конвеєра). The PipelineSession контекст успадковує Сесія SageMaker і реалізує зручні методи для взаємодії з іншими об’єктами та ресурсами SageMaker, такими як навчальні завдання, кінцеві точки, вхідні набори даних у Служба простого зберігання Amazon (Amazon S3) і так далі. Визначаючи конвеєри SageMaker, слід використовувати PipelineSession протягом звичайної сесії 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,
)

Запускайте конвеєри в локальному режимі для економічно ефективних і швидких ітерацій під час розробки

Ви можете запустити трубопроводу в локальному режимі використання LocalPipelineSession контекст. У цьому режимі конвеєр і завдання виконуються локально, використовуючи ресурси на локальній машині, замість ресурсів, керованих SageMaker. Локальний режим забезпечує економічно ефективний спосіб ітерації коду конвеєра з меншою підмножиною даних. Після локального тестування конвеєра його можна масштабувати для запуску за допомогою PipelineSession контексті.

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

Керуйте конвеєром SageMaker через версії

Версійність артефактів і визначень конвеєрів є загальною вимогою життєвого циклу розробки. Ви можете створити кілька версій конвеєра, назвавши об’єкти конвеєра унікальним префіксом або суфіксом, найпоширенішим з яких є позначка часу, як показано в наступному коді:

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

Організуйте та відстежуйте конвеєрні роботи SageMaker за допомогою інтеграції з SageMaker Experiments

SageMaker Pipelines можна легко інтегрувати Експерименти SageMaker для організації та відстеження прогонів трубопроводів. Це досягається специфікацією PipelineExperimentConfig на момент створення a об'єкт трубопроводу. За допомогою цього об’єкта конфігурації можна вказати назву експерименту та назву випробування. Деталі виконання конвеєра SageMaker організовуються відповідно до вказаного експерименту та випробування. Якщо ви явно не вказуєте назву експерименту, для назви експерименту використовується назва конвеєра. Подібним чином, якщо ви явно не вкажете ім’я пробної версії, для назви пробної групи або назви групи запуску використовується ідентифікатор конвеєрного циклу. Перегляньте наступний код:

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

Безпечно запускайте конвеєри SageMaker у приватному VPC

Щоб убезпечити робочі навантаження ML, найкращою практикою є розгортання завдань, організованих SageMaker Pipelines, у безпечній мережевій конфігурації в межах приватного VPC, приватних підмереж і груп безпеки. Щоб забезпечити та забезпечити використання цього безпечного середовища, ви можете реалізувати наступне Управління ідентифікацією та доступом AWS (IAM) політика для Роль виконання SageMaker (це роль, яку виконує конвеєр під час його роботи). Ви також можете додати політику для запуску завдань, організованих SageMaker Pipelines, у режимі ізоляції мережі.

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

Для прикладу впровадження конвеєра з цими засобами контролю безпеки див Організація завдань, реєстрація моделі та безперервне розгортання за допомогою Amazon SageMaker у безпечному середовищі.

Відстежуйте вартість конвеєрів за допомогою тегів

Використання самих конвеєрів SageMaker безкоштовне; ви платите за обчислювальні ресурси та ресурси зберігання, які ви використовуєте в рамках окремих етапів конвеєра, таких як обробка, навчання та пакетний аналіз. Щоб агрегувати витрати на прогін трубопроводу, ви можете включити теги на кожному кроці конвеєра, який створює ресурс. Потім на ці теги можна посилатися в провіднику витрат, щоб відфільтрувати та агрегувати загальну вартість роботи конвеєра, як показано в наступному прикладі:

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

У провіднику витрат тепер можна отримати вартість, відфільтровану за тегом:

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

Шаблони проектування для деяких типових сценаріїв

У цьому розділі ми обговорюємо шаблони проектування для деяких типових випадків використання конвеєрів SageMaker.

Запустіть легку функцію Python за допомогою кроку Lambda

Функції Python повсюдно присутні в робочих процесах ML; вони використовуються в попередній обробці, постобробці, оцінці тощо. Lambda — це безсерверний обчислювальний сервіс, який дозволяє запускати код без підготовки та керування серверами. За допомогою Lambda ви можете запускати код на бажаній мові, яка включає Python. Ви можете використовувати це для запуску спеціального коду Python як частини вашого конвеєра. Крок Лямбда дозволяє запускати функції Lambda як частину конвеєра SageMaker. Почніть із наступного коду:

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

Створіть лямбда-функцію за допомогою Помічник лямбда 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")

Виклик кроку Лямбда:

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

Передача даних між кроками

Вхідними даними для кроку конвеєра є або доступне розташування даних, або дані, згенеровані одним із попередніх кроків конвеєра. Ви можете надати цю інформацію як a ProcessingInput параметр. Давайте розглянемо кілька сценаріїв використання ProcessingInput.

Сценарій 1: передати вихідні дані (примітивні типи даних) лямбда-кроку до етапу обробки

Примітивні типи даних відносяться до скалярних типів даних, таких як рядок, ціле число, логічний і плаваючий.

Наступний фрагмент коду визначає функцію Lambda, яка повертає словник змінних із простими типами даних. Ваш код функції Lambda поверне JSON пар ключ-значення під час виклику з кроку Lambda в конвеєрі SageMaker.

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

У визначенні конвеєра ви можете визначити параметри конвеєра SageMaker, які належать до певного типу даних, і встановити змінну на вивід функції 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,
)

Сценарій 2: передайте вихідні дані (непримітивні типи даних) лямбда-кроку на етап обробки

Непримітивні типи даних стосуються нескалярних типів даних (наприклад, NamedTuple). У вас може виникнути сценарій, коли вам доведеться повернути непримітивний тип даних із функції лямбда. Для цього вам потрібно перетворити ваш непримітивний тип даних на рядок:

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

Потім ви можете використовувати цей рядок як вхідні дані для наступного кроку в конвеєрі. Щоб використати названий кортеж у коді, використовуйте eval() щоб розібрати вираз Python у рядку:

# 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: передайте вихідні дані кроку через файл властивостей

Ви також можете зберігати вихідні дані кроку обробки в a файл властивостей JSON для подальшого споживання в a ConditionStep або інший ProcessingStep. Ви можете використовувати Функція JSONGet запитувати a файл властивостей. Дивіться наступний код:

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

Припустімо, що вміст файлу властивостей такий:

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

У цьому випадку його можна запитати для певного значення та використовувати на наступних кроках за допомогою функції JsonGet:

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

Параметризуйте змінну у визначенні конвеєра

Часто бажано параметризувати змінні, щоб їх можна було використовувати під час виконання, наприклад, для створення S3 URI. Ви можете параметризувати рядок так, щоб він оцінювався під час виконання за допомогою Join функція. У наступному фрагменті коду показано, як визначити змінну за допомогою Join і використовуйте її, щоб установити місце виведення на етапі обробки:

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

Запустіть паралельний код над iterable

Деякі робочі процеси ML виконують код у паралельних циклах for над статичним набором елементів (an ітерабельний). Це може бути один і той самий код, який виконується на різних даних, або інший фрагмент коду, який потрібно запускати для кожного елемента. Наприклад, якщо у вас дуже велика кількість рядків у файлі і ви хочете пришвидшити час обробки, ви можете покластися на попередній шаблон. Якщо ви хочете виконати різні перетворення для певних підгруп у даних, вам, можливо, доведеться запустити окремий фрагмент коду для кожної підгрупи в даних. Наступні два сценарії ілюструють, як ви можете розробити конвеєри SageMaker для цієї мети.

Сценарій 1: реалізуйте логіку обробки різних частин даних

Ви можете запустити завдання обробки з кількома екземплярами (встановивши instance_count до значення, більшого за 1). Це розподіляє вхідні дані з Amazon S3 на всі екземпляри обробки. Потім ви можете використовувати сценарій (process.py) для роботи з певною частиною даних на основі номера екземпляра та відповідного елемента в списку елементів. Логіку програмування в process.py можна записати таким чином, щоб інший модуль або фрагмент коду запускався залежно від списку елементів, які він обробляє. У наступному прикладі визначено процесор, який можна використовувати в 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. Виконайте послідовність кроків

Якщо у вас є послідовність кроків, які потрібно виконати паралельно, ви можете визначити кожну послідовність як незалежний конвеєр SageMaker. Потім запуск цих конвеєрів SageMaker можна запустити з функції Lambda, яка є частиною LambdaStep у батьківському конвеєрі. Наступний фрагмент коду ілюструє сценарій, у якому запускаються два різні конвеєри 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

Висновок

У цій публікації ми обговорили деякі найкращі методи ефективного використання та обслуговування конвеєрів SageMaker. Ми також надали певні шаблони, які ви можете прийняти під час проектування робочих процесів за допомогою конвеєрів SageMaker, незалежно від того, створюєте ви нові конвеєри чи переміщуєте робочі процеси ML з інших інструментів оркестровки. Щоб розпочати роботу з SageMaker Pipelines для оркестровки робочого процесу ML, див зразки коду на GitHub та Трубопроводи для створення моделей Amazon SageMaker.

Про авторів

Pinak Panigrahi працює з клієнтами над створенням рішень на основі машинного навчання для вирішення стратегічних бізнес-завдань на AWS. Коли він не займається машинним навчанням, його можна знайти в поході, читанні книги або перегляді спортивних змагань.

Мінакшісундарам Тандавараян працює в AWS спеціалістом зі штучного інтелекту та машинного навчання. Він має пристрасть до проектування, створення та просування орієнтованих на людину даних і аналітики. Meena зосереджується на розробці стійких систем, які забезпечують вимірні конкурентні переваги для стратегічних клієнтів AWS. Міна — спеціаліст, який мислить у сфері дизайну та прагне підштовхнути бізнес до нових способів роботи через інновації, інкубацію та демократизацію.

• Розповсюдження контенту та PR на основі SEO. Отримайте посилення сьогодні.
• PlatoData.Network Vertical Generative Ai. Додайте собі сили. Доступ тут.
• PlatoAiStream. Web3 Intelligence. Розширення знань. Доступ тут.
• ПлатонЕСГ. Автомобільні / електромобілі, вуглець, CleanTech, Енергія, Навколишнє середовище, Сонячна, Поводження з відходами. Доступ тут.
• PlatoHealth. Розвідка про біотехнології та клінічні випробування. Доступ тут.
• ChartPrime. Розвивайте свою торгову гру за допомогою ChartPrime. Доступ тут.
• BlockOffsets. Модернізація екологічної компенсаційної власності. Доступ тут.
• джерело: https://aws.amazon.com/blogs/machine-learning/best-practices-and-design-patterns-for-building-machine-learning-workflows-with-amazon-sagemaker-pipelines/