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

Amazon SageMaker is a fully managed machine learning service that enables data scientists and developers to build, train, and deploy ML models at scale. This tutorial covers the complete ML lifecycle on AWS.

Why AWS SageMaker?

SageMaker Advantages:

• Fully managed: No infrastructure to manage
• End-to-end: Complete ML lifecycle support
• Scalable: Train on any scale with managed infrastructure
• Integrated: Native AWS service integration
• Cost-effective: Pay only for what you use

Key Components:

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

Prerequisites

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

Configure AWS credentials
aws configure
Enter: AWS Access Key ID, Secret Access Key, Region (e.g., us-east-1)

Quick Start

1. Setup SageMaker Session

import boto3
import sagemaker
from sagemaker import getexecutionrole

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

role = getexecutionrole()  # Or specify IAM role ARN


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

2. Prepare Training Data

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)


Save to 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"Training data: {trainpath}")
print(f"Test data: {testpath}")

Built-in Algorithms

1. XGBoost Training

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

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

Create 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
    }
)

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

)

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

2. Linear Learner

from sagemaker import LinearLearner

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

    predictortype="multiclassclassifier",

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

)

Prepare data in RecordIO format
trainrecords = linear.recordset(

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

    channel='train'
)

Train
linear.fit(trainrecords)

Custom Training Scripts

1. Scikit-learn Training

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


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

        maxdepth=args.maxdepth

    )
    model.fit(X, y)

    # Evaluate
    accuracy = accuracyscore(y, model.predict(X))
    print(f"Training accuracy: {accuracy}")

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


if name == 'main':
    main()

# Run 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. PyTorch Training

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

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

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

    # Train
    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}")


    # Save 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})

Model Deployment

1. Real-time Endpoint

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

    endpointname="iris-xgboost-endpoint"
)

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

print(f"Predictions: {predictions}")

Clean up
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

# Create transformer
transformer = xgbestimator.transformer(

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

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

Run batch transform
transformer.transform(
    data=testpath,

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

)

transformer.wait()

4. Multi-Model Endpoint

from sagemaker.multidatamodel import MultiDataModel

Create 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"

)

Add models dynamically
mme.addmodel(modeldatasource="s3://bucket/model1.tar.gz")
mme.addmodel(modeldatasource="s3://bucket/model2.tar.gz")


Predict with specific model
predictor.predict(data, targetmodel="model1.tar.gz")

Hyperparameter Tuning

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

Define hyperparameter ranges
hyperparameterranges = {

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

}

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

    hyperparameterranges=hyperparameterranges,

    maxjobs=10,
    maxparalleljobs=3
)

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

Get best model
bestestimator = tuner.bestestimator()

SageMaker Experiments

from sagemaker.experiments import Run

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

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

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

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

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

    run.logmetric("f1score", 0.94)


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

SageMaker Debugger

from sagemaker.debugger import Rule, ruleconfigs, DebuggerHookConfig

Configure debugger
debuggerhookconfig = DebuggerHookConfig(

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

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

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

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

Create estimator with 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

Register 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}")


Approve model
modelpackage.updateapprovalstatus(
    approvalstatus="Approved"

)

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

)

Cost Optimization

1. Spot Instances

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

    instancetype="ml.m5.xlarge",
    usespotinstances=True,
    maxwait=3600,  # Max wait time

    maxrun=3600,   # Max run time
    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  # Keep warm for 1 hour
)

Best Practices

1. Project Structure

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

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

2. Environment Variables

# Available in training container
os.environ['SMMODELDIR']      # /opt/ml/model
os.environ['SMCHANNELTRAIN']  # /opt/ml/input/data/train
os.environ['SMNUMGPUS']       # Number of GPUs
os.environ['SMNUMCPUS']       # Number of CPUs
os.environ['SMHOSTS']          # List of hosts (distributed)

os.environ['SMCURRENT_HOST']   # Current host name

Conclusion

AWS SageMaker provides comprehensive ML capabilities:

Built-in algorithms: XGBoost, Linear Learner, etc.

Custom training: Bring your own scripts

Flexible deployment: Real-time, serverless, batch

Experiment tracking: Compare and reproduce

Cost optimization: Spot instances, warm pools

Key takeaways:

• Use built-in algorithms for quick starts
• Leverage hyperparameter tuning for optimization
• Implement proper model registry workflow
• Use spot instances to reduce costs
• Monitor models in production