A collection of notes and resources
既存のアーキテクチャと要件に基づいてデプロイインフラストラクチャを選択する
| エンドポイントタイプ | パフォーマンス | コスト | レイテンシー | ユースケース |
|---|---|---|---|---|
| リアルタイム | 高 | 高 | 低（ms） | Webアプリ、リアルタイム推論 |
| サーバーレス | 中 | 低～中 | 中（秒） | 不定期なトラフィック |
| 非同期 | 中 | 中 | 高（分） | 大きなペイロード、長時間処理 |
| バッチ | 高（一括） | 低 | N/A | 定期的な大量処理 |
# GPU インスタンスの選択例
estimator = PyTorch(
entry_point='train.py',
role=role,
instance_count=4,
instance_type='ml.p4d.24xlarge', # 8x A100 GPU
framework_version='2.0'
)
# 推論最適化インスタンス
predictor = model.deploy(
initial_instance_count=1,
instance_type='ml.inf2.xlarge' # AWS Inferentia
)
from sagemaker.workflow.pipeline import Pipeline
from sagemaker.workflow.steps import ProcessingStep, TrainingStep
# パイプラインステップの定義
pipeline = Pipeline(
name='ml-pipeline',
steps=[
processing_step,
training_step,
evaluation_step
]
)
pipeline.upsert(role_arn=role)
from sagemaker.multidatamodel import MultiDataModel
# マルチモデルエンドポイント
mme = MultiDataModel(
name='multi-model',
model_data_prefix=f's3://{bucket}/models/',
image_uri=container,
role=role
)
predictor = mme.deploy(
initial_instance_count=1,
instance_type='ml.m5.xlarge'
)
# 特定モデルで推論
predictions = predictor.predict(data, target_model='model1.tar.gz')
from sagemaker.multidatamodel import MultiDataModel
from sagemaker.pipeline import PipelineModel
# 前処理 + モデル推論のパイプライン
pipeline_model = PipelineModel(
name='pipeline-model',
role=role,
models=[
preprocessing_model,
inference_model
]
)
predictor = pipeline_model.deploy(
initial_instance_count=1,
instance_type='ml.m5.xlarge'
)
# ブルー/グリーングデプロイ
predictor = model.deploy(
initial_instance_count=1,
instance_type='ml.m5.xlarge',
endpoint_name='production-endpoint'
)
# 新バージョンを追加
predictor.update_endpoint(
initial_instance_count=1,
instance_type='ml.m5.xlarge',
variant_name='green-variant',
initial_variant_weight=0 # 最初は0%
)
# トラフィックをシフト
predictor.update_endpoint_weights_and_capacities(
desired_weights_and_capacities=[
{'VariantName': 'blue-variant', 'DesiredWeight': 50},
{'VariantName': 'green-variant', 'DesiredWeight': 50}
]
)
from sagemaker.transformer import Transformer
transformer = Transformer(
model_name='my-model',
instance_count=1,
instance_type='ml.m5.xlarge',
output_path=f's3://{bucket}/batch-output/'
)
transformer.transform(
data=f's3://{bucket}/batch-input/',
content_type='text/csv',
split_type='Line'
)
トラフィックパターン
├─ 常時トラフィックあり？
│ ├─ Yes → レイテンシー要件は？
│ │ ├─ < 100ms → リアルタイムエンドポイント（常時稼働）
│ │ └─ > 1秒 → 非同期エンドポイント
│ └─ No → トラフィック予測可能？
│ ├─ Yes → サーバーレスエンドポイント
│ └─ No → バッチ推論
└─ 大量データの一括処理？
└─ Yes → バッチ Transform