Tutorial Lengkap AWS SageMaker: End-to-End ML Pipeline

Amazon SageMaker adalah layanan machine learning terkelola penuh yang memungkinkan data scientist dan developer membangun, melatih, dan deploy model ML dalam skala besar. Tutorial ini mencakup siklus ML lengkap di AWS.

Mengapa AWS SageMaker?

Keunggulan SageMaker:

• Fully managed: Tidak perlu mengelola infrastruktur
• End-to-end: Dukungan siklus ML lengkap
• Scalable: Latih dalam skala apapun dengan infrastruktur terkelola
• Terintegrasi: Integrasi native dengan layanan AWS
• Hemat biaya: Bayar hanya yang digunakan

Komponen Utama:

• SageMaker Studio (IDE)
• SageMaker Training
• SageMaker Inference
• SageMaker Pipelines
• SageMaker Feature Store
• SageMaker Model Monitor

Prerequisites

# Install AWS CLI dan SDK
pip install boto3 sagemaker pandas scikit-learn

Konfigurasi kredensial AWS
aws configure
Masukkan: AWS Access Key ID, Secret Access Key, Region (misal: us-east-1)

Quick Start

1. Setup SageMaker Session

import boto3
import sagemaker
from sagemaker import getexecutionrole

Buat session
session = sagemaker.Session()
bucket = session.defaultbucket()

role = getexecutionrole()  # Atau tentukan IAM role ARN


print(f"Bucket: {bucket}")
print(f"Role: {role}")
print(f"Region: {session.botoregionname}")

2. Siapkan Data Training

import pandas as pd
from sklearn.datasets import loadiris
from sklearn.modelselection import traintestsplit


Load sample data
iris = loadiris()
df = pd.DataFrame(iris.data, columns=iris.featurenames)

df['target'] = iris.target

Split data
traindf, testdf = traintestsplit(df, testsize=0.2, randomstate=42)


Simpan ke S3
trainpath = f"s3://{bucket}/iris/train/train.csv"
testpath = f"s3://{bucket}/iris/test/test.csv"


traindf.tocsv(trainpath, index=False)
testdf.tocsv(testpath, index=False)


print(f"Data training: {trainpath}")
print(f"Data test: {testpath}")

Algoritma Built-in

1. Training XGBoost

from sagemaker.estimator import Estimator
from sagemaker.inputs import TrainingInput

Dapatkan container XGBoost
container = sagemaker.imageuris.retrieve(
    framework="xgboost",
    region=session.botoregionname,
    version="1.5-1"
)

Buat estimator
xgbestimator = Estimator(

    imageuri=container,
    role=role,
    instancecount=1,

    instancetype="ml.m5.xlarge",
    outputpath=f"s3://{bucket}/iris/output",

    sagemakersession=session,
    hyperparameters={
        "objective": "multi:softmax",
        "numclass": 3,

        "numround": 100,
        "maxdepth": 5,

        "eta": 0.2
    }
)

Definisikan input training
traininput = TrainingInput(
    s3data=trainpath,
    contenttype="csv"

)

Latih model
xgbestimator.fit({"train": traininput})

2. Linear Learner

from sagemaker import LinearLearner

Buat estimator Linear Learner
linear = LinearLearner(
    role=role,
    instancecount=1,
    instancetype="ml.m5.large",

    predictortype="multiclassclassifier",

    numclasses=3,
    outputpath=f"s3://{bucket}/linear/output"

)

Siapkan data dalam format RecordIO
trainrecords = linear.recordset(

    traindf.drop('target', axis=1).values.astype('float32'),
    traindf['target'].values.astype('float32'),

    channel='train'
)

Latih
linear.fit(trainrecords)

Script Training Custom

1. Training Scikit-learn

# trainsklearn.py
import argparse
import joblib
import os
import pandas as pd
from sklearn.ensemble import RandomForestClassifier
from sklearn.metrics import accuracyscore

def parseargs():

    parser = argparse.ArgumentParser()
    parser.addargument('--n-estimators', type=int, default=100)
    parser.addargument('--max-depth', type=int, default=10)

    parser.addargument('--model-dir', type=str, default=os.environ.get('SMMODELDIR'))
    parser.addargument('--train', type=str, default=os.environ.get('SMCHANNELTRAIN'))

    return parser.parseargs()

def main():
    args = parseargs()


    # Load data
    traindf = pd.readcsv(os.path.join(args.train, 'train.csv'))

    X = traindf.drop('target', axis=1)
    y = traindf['target']


    # Latih model
    model = RandomForestClassifier(
        nestimators=args.nestimators,

        maxdepth=args.maxdepth

    )
    model.fit(X, y)

    # Evaluasi
    accuracy = accuracyscore(y, model.predict(X))
    print(f"Akurasi training: {accuracy}")

    # Simpan model
    joblib.dump(model, os.path.join(args.modeldir, 'model.joblib'))


if name == 'main':
    main()

# Jalankan training job
from sagemaker.sklearn import SKLearn

sklearnestimator = SKLearn(
    entrypoint="trainsklearn.py",
    role=role,
    instancecount=1,

    instancetype="ml.m5.large",
    frameworkversion="1.0-1",

    pyversion="py3",
    hyperparameters={
        "n-estimators": 200,
        "max-depth": 15
    }
)

sklearnestimator.fit({"train": trainpath})

2. Training PyTorch

# trainpytorch.py
import argparse
import os
import torch
import torch.nn as nn
import torch.optim as optim
from torch.utils.data import DataLoader, TensorDataset
import pandas as pd

class Net(nn.Module):
    def init(self, inputsize, numclasses):

        super(Net, self).init()
        self.fc1 = nn.Linear(inputsize, 64)
        self.fc2 = nn.Linear(64, 32)
        self.fc3 = nn.Linear(32, numclasses)


    def forward(self, x):
        x = torch.relu(self.fc1(x))
        x = torch.relu(self.fc2(x))
        return self.fc3(x)

def train(args):
    device = torch.device("cuda" if torch.cuda.isavailable() else "cpu")

    # Load data
    traindf = pd.readcsv(os.path.join(args.train, 'train.csv'))
    X = torch.tensor(traindf.drop('target', axis=1).values, dtype=torch.float32)

    y = torch.tensor(traindf['target'].values, dtype=torch.long)

    dataset = TensorDataset(X, y)
    loader = DataLoader(dataset, batchsize=args.batchsize, shuffle=True)

    # Buat model
    model = Net(X.shape[1], args.numclasses).to(device)

    criterion = nn.CrossEntropyLoss()
    optimizer = optim.Adam(model.parameters(), lr=args.lr)

    # Latih
    for epoch in range(args.epochs):
        model.train()
        totalloss = 0
        for batchX, batchy in loader:
            batchX, batchy = batchX.to(device), batchy.to(device)

            optimizer.zerograd()

            outputs = model(batchX)
            loss = criterion(outputs, batchy)

            loss.backward()
            optimizer.step()
            totalloss += loss.item()

        print(f"Epoch {epoch+1}/{args.epochs}, Loss: {totalloss/len(loader):.4f}")


    # Simpan model
    torch.save(model.statedict(), os.path.join(args.modeldir, 'model.pth'))


if name == 'main':
    parser = argparse.ArgumentParser()
    parser.addargument('--epochs', type=int, default=10)
    parser.addargument('--batch-size', type=int, default=32)

    parser.addargument('--lr', type=float, default=0.001)
    parser.addargument('--num-classes', type=int, default=3)

    parser.addargument('--model-dir', type=str, default=os.environ.get('SMMODELDIR'))
    parser.addargument('--train', type=str, default=os.environ.get('SMCHANNELTRAIN'))

    args = parser.parseargs()
    train(args)

from sagemaker.pytorch import PyTorch

pytorchestimator = PyTorch(

    entrypoint="trainpytorch.py",

    role=role,
    instancecount=1,
    instancetype="ml.m5.large",

    frameworkversion="1.13",
    pyversion="py39",

    hyperparameters={
        "epochs": 20,
        "batch-size": 32,
        "lr": 0.001,
        "num-classes": 3
    }
)

pytorchestimator.fit({"train": trainpath})

Deployment Model

1. Real-time Endpoint

# Deploy model
predictor = xgbestimator.deploy(
    initialinstancecount=1,
    instancetype="ml.m5.large",

    endpointname="iris-xgboost-endpoint"
)

Buat prediksi
import numpy as np
testdata = testdf.drop('target', axis=1).values[:5]
predictions = predictor.predict(testdata)

print(f"Prediksi: {predictions}")

Bersihkan
predictor.deleteendpoint()

2. Serverless Inference

from sagemaker.serverless import ServerlessInferenceConfig

serverlessconfig = ServerlessInferenceConfig(

    memorysizeinmb=2048,
    maxconcurrency=10

)

predictor = xgbestimator.deploy(
    serverlessinferenceconfig=serverlessconfig,

    endpointname="iris-serverless-endpoint"
)

3. Batch Transform

# Buat transformer
transformer = xgbestimator.transformer(

    instancecount=1,
    instancetype="ml.m5.large",

    outputpath=f"s3://{bucket}/iris/batch-output"
)

Jalankan batch transform
transformer.transform(
    data=testpath,

    contenttype="text/csv",
    splittype="Line"

)

transformer.wait()

4. Multi-Model Endpoint

from sagemaker.multidatamodel import MultiDataModel

Buat multi-model endpoint
mme = MultiDataModel(
    name="multi-model-endpoint",
    modeldataprefix=f"s3://{bucket}/models/",

    imageuri=container,
    role=role
)

Deploy
predictor = mme.deploy(
    initialinstancecount=1,
    instancetype="ml.m5.large"

)

Tambah model secara dinamis
mme.addmodel(modeldatasource="s3://bucket/model1.tar.gz")
mme.addmodel(modeldatasource="s3://bucket/model2.tar.gz")


Prediksi dengan model spesifik
predictor.predict(data, targetmodel="model1.tar.gz")

Hyperparameter Tuning

from sagemaker.tuner import (
    HyperparameterTuner,
    IntegerParameter,
    ContinuousParameter
)

Definisikan range hyperparameter
hyperparameterranges = {

    "maxdepth": IntegerParameter(3, 10),
    "eta": ContinuousParameter(0.1, 0.5),
    "numround": IntegerParameter(50, 200)

}

Buat tuner
tuner = HyperparameterTuner(
    estimator=xgbestimator,
    objectivemetricname="validation:merror",
    objectivetype="Minimize",

    hyperparameterranges=hyperparameterranges,

    maxjobs=10,
    maxparalleljobs=3
)

Jalankan tuning job
tuner.fit({"train": traininput, "validation": testinput})

Dapatkan model terbaik
bestestimator = tuner.bestestimator()

SageMaker Experiments

from sagemaker.experiments import Run

Buat experiment run
with Run(
    experimentname="iris-classification",

    runname="xgboost-run-1",
    sagemakersession=session

) as run:
    # Log parameter
    run.logparameter("algorithm", "xgboost")
    run.logparameter("maxdepth", 5)
    run.logparameter("numround", 100)

    # Latih model
    xgbestimator.fit({"train": traininput})

    # Log metrik
    run.logmetric("accuracy", 0.95)

    run.logmetric("f1score", 0.94)


    # Log artifact
    run.logartifact(name="model", value=xgbestimator.modeldata)

SageMaker Debugger

from sagemaker.debugger import Rule, ruleconfigs, DebuggerHookConfig

Konfigurasi debugger
debuggerhookconfig = DebuggerHookConfig(

    s3outputpath=f"s3://{bucket}/debug-output",

    collectionconfigs=[
        CollectionConfig(name="metrics"),
        CollectionConfig(name="losses")
    ]
)

Tambah debugging rules
rules = [
    Rule.sagemaker(ruleconfigs.lossnotdecreasing()),

    Rule.sagemaker(ruleconfigs.overfit()),
    Rule.sagemaker(ruleconfigs.vanishinggradient())
]

Buat estimator dengan debugger
estimator = PyTorch(
    entrypoint="train.py",

    role=role,
    instancecount=1,
    instancetype="ml.p3.2xlarge",

    frameworkversion="1.13",
    debuggerhookconfig=debuggerhookconfig,
    rules=rules
)

Model Registry

from sagemaker.model import Model
from sagemaker import ModelPackage

Daftarkan model
modelpackage = xgbestimator.register(
    modelpackagegroupname="iris-models",

    contenttypes=["text/csv"],
    responsetypes=["text/csv"],

    inferenceinstances=["ml.m5.large"],
    transforminstances=["ml.m5.large"],

    approvalstatus="PendingManualApproval"
)

print(f"Model ARN: {modelpackage.modelpackagearn}")


Setujui model
modelpackage.updateapprovalstatus(
    approvalstatus="Approved"

)

Deploy dari registry
model = ModelPackage(
    role=role,
    modelpackagearn=modelpackage.modelpackagearn
)
predictor = model.deploy(
    initialinstancecount=1,
    instancetype="ml.m5.large"

)

Optimasi Biaya

1. Spot Instances

estimator = Estimator(
    imageuri=container,
    role=role,
    instancecount=1,

    instancetype="ml.m5.xlarge",
    usespotinstances=True,
    maxwait=3600,  # Waktu tunggu maksimal

    maxrun=3600,   # Waktu run maksimal
    checkpoints3uri=f"s3://{bucket}/checkpoints"
)

2. Managed Warm Pools

from sagemaker.instancegroup import InstanceGroup

estimator = Estimator(
    imageuri=container,
    role=role,
    instancecount=1,

    instancetype="ml.m5.xlarge",
    keepaliveperiodinseconds=3600  # Tetap warm selama 1 jam
)

Best Practices

1. Struktur Project

sagemaker-project/
├── src/
│   ├── train.py
│   ├── inference.py
│   └── preprocess.py
├── notebooks/
│   └── exploration.ipynb
├── tests/
│   └── testtrain.py

├── pipelines/
│   └── trainingpipeline.py
└── requirements.txt

2. Environment Variables

# Tersedia di container training
os.environ['SMMODELDIR']      # /opt/ml/model
os.environ['SMCHANNELTRAIN']  # /opt/ml/input/data/train
os.environ['SMNUMGPUS']       # Jumlah GPU
os.environ['SMNUMCPUS']       # Jumlah CPU
os.environ['SMHOSTS']          # Daftar hosts (distributed)

os.environ['SMCURRENT_HOST']   # Nama host saat ini

Kesimpulan

AWS SageMaker menyediakan kemampuan ML komprehensif:

Algoritma built-in: XGBoost, Linear Learner, dll.

Training custom: Bawa script sendiri

Deployment fleksibel: Real-time, serverless, batch

Experiment tracking: Bandingkan dan reproduksi

Optimasi biaya: Spot instances, warm pools

Key takeaways:

• Gunakan algoritma built-in untuk quick start
• Manfaatkan hyperparameter tuning untuk optimasi
• Implementasikan workflow model registry yang proper
• Gunakan spot instances untuk mengurangi biaya
• Monitor model di production