2024-05-17|閱讀時間 ‧ 約 29 分鐘

SageMaker-建構Pipelines真的有必要嗎?

一.引言

  除了上篇提到的 Data Wrangler 外,SageMaker 還有許多配套的功能,其中有個叫做 Pipelines 的東西,說是可以用來構建、 管理及自動化深度學習流程,能夠節省人工操作,有那麼神?這次就來試試 Pipelines 能夠為我們帶來什麼體驗。

二.Pipelines 介紹

  SageMaker Pipelines 提供了一個框架,讓你定義和自動化機器學習工作流程中的所有步驟(Step),包括資料預處理、特徵工程、模型訓練、模型評估和模型部署。 每個步驟可以由不同的SageMaker元件組成。

Step 為 Pipeline 中的基本單位,每個 Step 表示一個特定的任務或操作 :

  • ProcessingStep:用於資料處理和特徵工程
  • TrainingStep:用於模型訓練
  • TransformStep:用於批量推理
  • TuningStep:用於超參數調優
  • ConditionStep:用於條件判斷
  • CallbackStep:用於呼叫外部系統或服務
  • LambdaStep:用於執行自訂邏輯

接下來便簡易搭建一個 Pipeline 流程 :

1.準備環境

import sagemaker
from sagemaker.workflow.pipeline import Pipeline
from sagemaker.workflow.steps import ProcessingStep, TrainingStep
from sagemaker.processing import ScriptProcessor
from sagemaker.estimator import Estimator
from sagemaker.workflow.parameters import ParameterString
from sagemaker.workflow.pipeline_context import PipelineSession

# SageMaker 會話和腳色
sagemaker_session = sagemaker.Session()
role = 'your-iam-role'
bucket = 'your-s3-bucket'
prefix = 'your-data-prefix'
# Pipeline 會話
pipeline_session = PipelineSession()

2.定義參數

# 定義數據​前處理用執行個體
processing_instance_type = ParameterString(name="ProcessingInstanceType",
default_value="ml.m5.xlarge")
# 定義訓練用執行個體​
training_instance_type = ParameterString(name="TrainingInstanceType",
default_value="ml.p3.2xlarge")

3.定義數據前處理step

# 定義數據處理流程
processor = ScriptProcessor(
role=role,
image_uri='your-processing-container',
command=['python3'],
instance_count=1,
instance_type=processing_instance_type,
sagemaker_session=sagemaker_session
)
# 定義 Step​
step_process = ProcessingStep(
name='DataProcessing',
processor=processor,
inputs=[sagemaker.processing.ProcessingInput(
source=f's3://{bucket}/{prefix}/raw-data',
destination='/opt/ml/processing/input'
)],
outputs=[sagemaker.processing.ProcessingOutput(
source='/opt/ml/processing/output',
destination=f's3://{bucket}/{prefix}/processed-data'
)],
code='processing_script.py'
)

4.定義訓練Step

# 定義訓練流程
estimator = PyTorch(
entry_point='train.py',
role=role,
instance_count=1,
instance_type=training_instance_type,
framework_version='1.8.0',
py_version='py37',
script_mode=True,
output_path='s3://your-bucket/your-prefix/model',
sagemaker_session=sagemaker_session
)
# 定義訓練Step​
step_train = TrainingStep(
name='ModelTraining',
estimator=estimator,
inputs={'train': 's3://your-bucket/your-prefix/processed-data'}
)

5.定義模型部屬Step

model = Model(
image_uri='your-inference-container',
model_data=step_train.properties.ModelArtifacts.S3ModelArtifacts,
role=role
)

step_model = ModelStep(
name='ModelDeployment',
model=model,
instance_count=1,
instance_type='ml.m5.large'
)

6.定義及創建Pipeline

# 定义Pipeline
pipeline = Pipeline(
name='MyPipeline',
parameters=[processing_instance_type, training_instance_type],
steps=[step_process, step_train, step_model]
)

# 创建和启动Pipeline
pipeline.upsert(role=role)
pipeline.start()

  以上程式碼定義了三個 Step,分別包含數據前處理、模型訓練、模型部屬,並交由Pipeline 去順序執行,從示例可以看到,我們可以針對不同 Step 指定不同的執行個體,這意味著我們可以根據不同 Step 的運算需求使用最適合的環境,並且因為分離成不同的 Step,在維護及管理上便可以視為單一獨立的區塊去處理,在工作細化及分工上都可以帶來幫助。

三.總結

  SageMaker Pipeline 將整個訓練流程細分成多個 Step ,雖然增加了分工及管理上的優點,但前提是有著良好的分工狀態,若是專案不夠龐大,需要的處理不夠複雜,還是將所有流程寫在同一份程式碼內會相對好處理很多。

所以統整下大概符合下列條件再建構 Pipeline 才會有比較好的體驗 :

  • 資料量大且處理複雜:每天產生大量數據,需要多階段處理和特徵工程
  • 超參數調優和模型選擇:需要並行訓練多個模型並自動選擇最佳模型
  • 持續整合和部署:頻繁更新資料和模型,需要自動化CI/CD流程
  • 跨團隊協作和大規模管理:多個團隊協作,定義和管理複雜的工作流程
分享至
成為作者繼續創作的動力吧!
© 2024 vocus All rights reserved.