☁️
클라우드
SageMaker
Amazon SageMaker
AWS의 완전 관리형 머신러닝 플랫폼. 데이터 준비, 모델 학습, 배포, 모니터링까지 MLOps 전체 파이프라인을 지원하며, JumpStart로 파운데이션 모델도 활용할 수 있습니다.
📖 상세 설명
Amazon SageMaker는 2017년 AWS re:Invent에서 처음 발표된 완전 관리형 머신러닝 서비스입니다. 개발자와 데이터 과학자가 ML 모델을 빠르게 구축, 학습, 배포할 수 있도록 설계되었습니다. AWS 생태계와의 깊은 통합이 강점으로, S3, Lambda, Step Functions, CloudWatch 등과 자연스럽게 연동됩니다.
SageMaker의 핵심 구성요소는 다양합니다. Studio는 웹 기반 IDE로 노트북, 실험 관리, 모델 디버깅을 통합 제공합니다. Processing은 대규모 데이터 전처리를 분산 실행합니다. Training은 다양한 인스턴스 타입에서 학습을 실행하며, Spot Instance를 활용해 비용을 절감할 수 있습니다. Inference는 실시간 엔드포인트, 배치 변환, 서버리스 추론을 지원합니다.
최근에는 생성형 AI 지원이 대폭 강화되었습니다. JumpStart는 Llama, Falcon, FLAN-T5 등 오픈소스 파운데이션 모델을 원클릭으로 배포하고, 자체 데이터로 파인튜닝할 수 있습니다. Bedrock과 연동하면 Claude, Titan 등 관리형 LLM API도 사용 가능합니다.
MLOps 측면에서 SageMaker Pipelines는 ML 워크플로우를 코드로 정의하고 자동화합니다. Model Registry는 모델 버전을 관리하고, Model Monitor는 배포된 모델의 데이터 드리프트와 성능 저하를 모니터링합니다. Feature Store는 학습과 추론에 사용되는 피처를 중앙에서 관리합니다.
💻 코드 예제
# Amazon SageMaker를 활용한 ML 파이프라인 예제
import sagemaker
from sagemaker import get_execution_role
from sagemaker.estimator import Estimator
from sagemaker.processing import ScriptProcessor
from sagemaker.workflow.pipeline import Pipeline
from sagemaker.workflow.steps import ProcessingStep, TrainingStep, CreateModelStep
from sagemaker.workflow.parameters import ParameterString
from sagemaker.model import Model
from sagemaker.predictor import Predictor
# 세션 초기화
session = sagemaker.Session()
role = get_execution_role()
bucket = session.default_bucket()
# ==========================================
# 1. 데이터 전처리 (Processing)
# ==========================================
from sagemaker.sklearn.processing import SKLearnProcessor
sklearn_processor = SKLearnProcessor(
framework_version="1.2-1",
role=role,
instance_type="ml.m5.xlarge",
instance_count=1
)
# 전처리 실행
sklearn_processor.run(
code="preprocessing.py",
inputs=[
sagemaker.processing.ProcessingInput(
source="s3://my-bucket/raw-data/",
destination="/opt/ml/processing/input"
)
],
outputs=[
sagemaker.processing.ProcessingOutput(
output_name="train",
source="/opt/ml/processing/output/train",
destination=f"s3://{bucket}/processed/train"
),
sagemaker.processing.ProcessingOutput(
output_name="test",
source="/opt/ml/processing/output/test",
destination=f"s3://{bucket}/processed/test"
)
]
)
# ==========================================
# 2. 모델 학습 (Training)
# ==========================================
# PyTorch Estimator 예제
from sagemaker.pytorch import PyTorch
pytorch_estimator = PyTorch(
entry_point="train.py",
source_dir="./code",
role=role,
instance_count=1,
instance_type="ml.p3.2xlarge",
framework_version="2.1",
py_version="py310",
hyperparameters={
"epochs": 10,
"batch-size": 32,
"learning-rate": 0.001
},
use_spot_instances=True, # Spot 인스턴스로 비용 절감
max_run=3600,
max_wait=7200
)
# 학습 실행
pytorch_estimator.fit({
"train": f"s3://{bucket}/processed/train",
"test": f"s3://{bucket}/processed/test"
})
# ==========================================
# 3. 모델 배포 (Real-time Inference)
# ==========================================
predictor = pytorch_estimator.deploy(
initial_instance_count=1,
instance_type="ml.g4dn.xlarge",
endpoint_name="my-pytorch-endpoint"
)
# 추론 실행
import numpy as np
result = predictor.predict(np.array([[1.0, 2.0, 3.0, 4.0]]))
print(f"Prediction: {result}")
# ==========================================
# 4. SageMaker Pipelines (MLOps)
# ==========================================
from sagemaker.workflow.pipeline_context import PipelineSession
pipeline_session = PipelineSession()
# 파이프라인 파라미터
input_data = ParameterString(name="InputData", default_value=f"s3://{bucket}/raw-data")
instance_type = ParameterString(name="TrainingInstance", default_value="ml.m5.xlarge")
# 전처리 단계
processing_step = ProcessingStep(
name="DataPreprocessing",
processor=sklearn_processor,
inputs=[
sagemaker.processing.ProcessingInput(
source=input_data,
destination="/opt/ml/processing/input"
)
],
outputs=[
sagemaker.processing.ProcessingOutput(
output_name="train",
source="/opt/ml/processing/output/train"
)
],
code="preprocessing.py"
)
# 학습 단계
training_step = TrainingStep(
name="ModelTraining",
estimator=pytorch_estimator,
inputs={
"train": sagemaker.inputs.TrainingInput(
s3_data=processing_step.properties.ProcessingOutputConfig.Outputs["train"].S3Output.S3Uri
)
}
)
# 파이프라인 정의
pipeline = Pipeline(
name="ml-training-pipeline",
parameters=[input_data, instance_type],
steps=[processing_step, training_step]
)
# 파이프라인 생성 및 실행
pipeline.upsert(role_arn=role)
execution = pipeline.start()
# ==========================================
# 5. JumpStart로 LLM 배포
# ==========================================
from sagemaker.jumpstart.model import JumpStartModel
# Llama 2 모델 배포
llm_model = JumpStartModel(
model_id="meta-textgeneration-llama-2-7b",
role=role
)
llm_predictor = llm_model.deploy(
initial_instance_count=1,
instance_type="ml.g5.2xlarge",
endpoint_name="llama2-endpoint"
)
# LLM 추론
response = llm_predictor.predict({
"inputs": "What is machine learning?",
"parameters": {
"max_new_tokens": 256,
"temperature": 0.7
}
})
print(response)🗣️ 실무에서 이렇게 말해요
ML 리드: "학습 시간이 너무 오래 걸려요. p3.8xlarge로 올려볼까요?"
엔지니어: "그 전에 Spot Instance 먼저 적용해보죠. max_wait 설정하면 중단되도 체크포인트에서 재개할 수 있어요. 비용 70%까지 절감됩니다."
ML 리드: "좋아요. 그리고 SageMaker Debugger로 GPU 사용률도 확인해봐요. 배치 사이즈 문제일 수도 있어요."
면접관: "SageMaker에서 모델 배포 옵션의 차이점을 설명해주세요."
지원자: "Real-time Inference는 밀리초 단위 응답이 필요한 API용이고, Batch Transform은 대량 데이터 일괄 처리에 적합합니다. Serverless Inference는 트래픽이 불규칙하고 콜드스타트를 감수할 수 있을 때 비용 효율적이고, Async Inference는 응답 시간이 수 초~분 단위로 오래 걸리는 대형 모델에 사용합니다."
리뷰어: "엔드포인트 배포할 때 data_capture_config 빠졌어요. 프로덕션에서 Model Monitor 쓰려면 필수입니다."
작성자: "아, 추론 데이터 캡처 설정 추가하고, 베이스라인도 같이 생성해둘게요."
⚠️ 주의사항
- SageMaker 엔드포인트는 시간당 과금됩니다. 테스트 후 반드시 삭제하세요. predictor.delete_endpoint()를 호출하지 않으면 계속 비용이 발생합니다.
- 대형 모델(LLM 등) 학습 시 인스턴스 할당량(Quota)을 미리 확인하세요. p4d, p5 같은 고사양 인스턴스는 기본 할당량이 0인 경우가 많습니다.
- Spot Instance 사용 시 학습이 중단될 수 있습니다. 반드시 체크포인트를 S3에 저장하고, max_wait을 max_run보다 크게 설정하세요.