Mlops

Specialist guidance for MLOps on AWS. Covers platform selection, training job configuration, inference deployment patterns, CI/CD for ML, experiment tracking, model monitoring, and cost optimization.

Process

1. Identify the ML workload characteristics: model type (classical ML, deep learning, foundation model), training data volume, inference latency requirements, traffic pattern, team expertise 2. Use the awsknowledge MCP tools (mcp__plugin_aws-dev-toolkit_awsknowledge__aws___search_documentation, mcp__plugin_aws-dev-toolkit_awsknowledge__aws___read_documentation, mcp__plugin_aws-dev-toolkit_awsknowledge__aws___recommend) to verify current SageMaker instance types, limits, pricing, and feature availability 3. Select the appropriate MLOps platform using the decision matrix below 4. Design the training infrastructure (instance selection, distributed strategy, Spot configuration) 5. Design the inference topology (real-time, serverless, batch, async) 6. Configure the ML pipeline (SageMaker Pipelines, Step Functions, or CI/CD integration) 7. Set up experiment tracking (MLflow on SageMaker or SageMaker Experiments) 8. Configure model monitoring (data quality, model quality, bias drift, feature attribution drift) 9. Recommend cost optimization strategies (Spot training, Savings Plans, Inferentia/Trainium, right-sizing)

Platform Selection Decision Matrix

Training Instance Selection

Training Instances

Inference Instances

Default to Trainium/Inferentia When Possible

Always evaluate ml.trn1 for training and ml.inf2 for inference before selecting GPU instances. Trainium offers up to 50% cost savings for training and Inferentia2 offers 50%+ cost savings for inference on supported model architectures. The AWS Neuron SDK supports PyTorch and TensorFlow natively. Only fall back to GPU instances when the model architecture is not supported by the Neuron compiler (check the Neuron model support matrix) or when the team needs CUDA-specific libraries.

Inference Deployment Decision Matrix

Real-time Endpoint Patterns

• Single-model endpoint: One model per endpoint. Simplest. Use for most production deployments.
• Multi-model endpoint (MME): Thousands of models behind one endpoint, loaded on demand from S3. Use when you have many similar models (per-customer, per-region) and cannot justify an endpoint per model. Trade-off: first-request latency while loading a model.
• Multi-container endpoint: Up to 15 containers per endpoint, invoked individually or as a serial pipeline. Use for A/B testing different model versions or combining pre/post-processing with inference.
• Shadow testing: Route production traffic to both current and candidate models simultaneously. Compare metrics before promoting. Always use shadow testing before replacing a production model because it reveals performance differences under real traffic that offline evaluation cannot capture.

Auto-Scaling

Default to target-tracking scaling on SageMakerVariantInvocationsPerInstance because it automatically adjusts instance count based on actual request load without requiring manual threshold tuning.

Target value: start at 70% of the max RPS the instance can handle (benchmark first)
Scale-in cooldown: 300 seconds (prevent flapping)
Scale-out cooldown: 60 seconds (respond quickly to load spikes)

Use Inference Recommender before production deployment to benchmark instance types and find the optimal instance/model combination. It runs load tests and reports latency, throughput, and cost per inference, replacing guesswork with data.

SageMaker Pipelines

Pipeline Step Types

Model Registry

The Model Registry is the central artifact store for production ML. Always register models through the registry because it provides:

• Version tracking: Every model gets an immutable version number with metadata (metrics, parameters, data lineage)
• Approval workflows: Models must be explicitly approved (manual or automated) before deployment, preventing untested models from reaching production
• Lineage: Links model versions to the training job, dataset, pipeline execution, and code commit that produced them
• Cross-account deployment: Approved models can be deployed to staging/production accounts via resource policies

Pipeline Best Practices

• Parameterize everything: Instance types, data paths, hyperparameters, and thresholds should be pipeline parameters, not hardcoded values. This enables reuse across environments (dev/staging/prod) without code changes.
• Use Condition steps for quality gates: After training, compare the candidate model metric against a threshold. Only register and deploy if the metric passes. This prevents model regressions from reaching production.
• Cache pipeline steps: Enable step caching to skip unchanged steps on re-execution, reducing pipeline run time and cost.
• Trigger pipelines from CI/CD: Use CodePipeline or GitHub Actions to trigger SageMaker Pipelines on code merge, creating a full ML CI/CD loop.

MLflow on AWS

Managed MLflow on SageMaker (Recommended Default)

Use managed MLflow on SageMaker as the default experiment tracking solution because it requires zero infrastructure management, scales automatically, and synchronizes with SageMaker Model Registry automatically.

• MLflow Apps: Latest offering with faster startup, cross-account sharing, and automatic model registration
• MLflow Tracking Servers: Traditional MLflow with configurable compute and storage; each project can have its own server
• Artifacts stored in S3 (durable, shareable, versioned)
• Native integration with SageMaker training jobs — metrics logged automatically
• Models registered in MLflow automatically appear in SageMaker Model Registry
• Deploy MLflow models directly to SageMaker endpoints without custom containers

Self-Hosted MLflow on EKS

Only choose self-hosted MLflow when you need custom plugins, specific MLflow versions not yet supported by the managed service, or multi-cloud portability with a single MLflow backend.

• Deploy MLflow server as a Kubernetes Deployment on EKS
• Use Amazon RDS (PostgreSQL) as the metadata/backend store for durability and query performance
• Use S3 as the artifact store with a dedicated bucket and lifecycle policies
• Front with an ALB + Cognito or IAM for authentication
• Operational overhead: you own patching, scaling, backups, and availability

When to Choose MLflow over Native SageMaker Experiments

• Team has existing MLflow workflows and muscle memory
• Multi-cloud or hybrid-cloud requirement where portability matters
• Need for MLflow-specific features (Prompt Registry, advanced tracing for agentic workflows)
• Want a single UI for experiment comparison across SageMaker and non-SageMaker training runs

Model Monitoring

Four Monitoring Dimensions

Monitoring Setup

1. Enable Data Capture on the endpoint to log inputs and outputs to S3 (asynchronous, no performance impact on inference) 2. Create baselines from the training dataset using DefaultModelMonitor for data quality or ModelQualityMonitor for model quality 3. Schedule monitoring jobs — hourly for high-traffic endpoints, daily for moderate traffic 4. Configure CloudWatch alarms on monitoring violations to trigger SNS notifications 5. Automate retraining: Use EventBridge to trigger a SageMaker Pipeline re-execution when monitoring detects sustained drift

When to Retrain vs When to Investigate

• Retrain when data quality monitoring shows gradual statistical drift (feature distributions shifting over time) and the model's accuracy metrics are declining — this is expected model staleness
• Investigate first when monitoring shows sudden, sharp changes — this typically indicates an upstream data pipeline issue, a schema change, or a bug, not genuine drift; retraining on bad data makes things worse

Distributed Training

Data Parallel

Use data parallel training when the model fits in a single GPU's memory but training is slow due to dataset size. Each GPU processes a different data batch, gradients are synchronized across GPUs. SageMaker's distributed data parallelism (SMDDP) library optimizes AllReduce/AllGather operations for better inter-node communication.

Model Parallel

Use model parallel training when the model does not fit in a single GPU's memory (large language models, large vision transformers). SageMaker's model parallelism (SMP) library supports tensor parallelism, pipeline parallelism, and expert parallelism. Use EFA-enabled instances (ml.p4d, ml.p5, ml.trn1) for model parallel training because inter-node communication is the bottleneck and EFA provides 400-3200 Gbps networking.

Hyperparameter Tuning Strategies

Always use Bayesian optimization as the default because it typically finds better hyperparameters in fewer trials than random search, directly reducing training cost.

Cost Optimization

Managed Spot Training

Always use Managed Spot Training for training jobs because training is inherently fault-tolerant (checkpointing lets you resume from the last saved state) and Spot provides 60-90% savings over On-Demand. The only exception is ultra-time-sensitive training where any interruption is unacceptable (rare in practice).

• Enable with use_spot_instances=True in the Estimator
• Set max_wait to 2x the expected training time to allow for interruptions
• Enable checkpointing to S3 so training resumes from the last checkpoint, not from scratch
• SageMaker automatically handles Spot interruption, checkpoint save, and job restart

SageMaker Savings Plans

Commit to consistent SageMaker usage (measured in $/hour) for 1 or 3 years. Savings Plans cover Studio Notebooks, Processing, Training, Real-Time Inference, and Batch Transform. Up to 64% savings over On-Demand. Use for production inference endpoints that run continuously.

Serverless Inference for Sporadic Traffic

Use Serverless Inference instead of real-time endpoints when traffic is sporadic or unpredictable. Real-time endpoints charge per instance-hour even when idle; Serverless charges per request and per millisecond of compute. A real-time ml.m5.large endpoint costs ~$100/month idle. Serverless at 100 requests/day costs under $5/month.

Instance Right-Sizing with Inference Recommender

Run SageMaker Inference Recommender before deploying to production. It benchmarks your model across instance types and reports latency percentiles, throughput, and cost per inference. Teams that skip this step typically overprovision by 2-3x because they guess conservatively.

Trainium and Inferentia

Evaluate Trainium (ml.trn1) for training and Inferentia2 (ml.inf2) for inference on every ML project. For supported model architectures (most PyTorch and TensorFlow models), these custom silicon instances deliver 50%+ cost savings compared to GPU instances. The Neuron SDK compiles models for these chips with minimal code changes. Only skip when the model uses CUDA-specific operations that Neuron does not support.

Anti-Patterns

• Training on notebooks instead of training jobs. Notebook training is not reproducible, cannot use Spot instances (60-90% savings lost), cannot distribute across multiple instances, and produces no training job metadata for lineage tracking. Always convert notebook experiments to SageMaker Training Jobs for anything beyond initial prototyping.

• Skipping Model Registry. Without a registry, there is no version history, no approval workflow, no lineage from model to training data, and no clean rollback path. A bad model deployed without registry tracking requires manual forensics to identify what changed.

• Real-time endpoints for batch workloads. A real-time endpoint running 24/7 to process a nightly batch job wastes money on 23 hours of idle compute. Batch Transform provisions instances only for the job duration and terminates them automatically.

• Single large instance instead of distributed training. A single ml.p5.48xlarge costs more per hour than multiple smaller instances delivering equivalent total compute. Distributed training also provides fault tolerance — if one node fails, only that node's work is lost, not the entire job.

• No model monitoring after deployment. Without monitoring, model drift goes undetected. Predictions degrade silently, and the team only discovers the problem when business metrics drop — weeks or months later. Data quality monitoring catches drift within hours.

• On-Demand training instances by default. SageMaker Managed Spot Training saves 60-90% and handles interruptions automatically with checkpointing. Training jobs are inherently resumable, making them ideal Spot workloads. On-Demand should be the exception, not the default.

• Deploying directly to production without shadow testing. Shadow testing routes live traffic to both the current and candidate models, comparing predictions and latency in real-time. Without it, the only signal that a new model is worse comes from production users experiencing degraded results.

• Not using experiment tracking (MLflow or SageMaker Experiments). Without experiment tracking, it is impossible to reproduce a previous result, compare hyperparameter choices across runs, or explain why one model version outperformed another. This wastes compute re-running experiments that were already tried.

• Storing artifacts locally instead of S3. Local artifacts are not durable (instance termination deletes them), not shareable across team members, and break CI/CD pipelines that expect S3 paths. S3 provides versioning, cross-account access, and lifecycle management.

• Ignoring Trainium/Inferentia. ml.trn1 and ml.inf2 instances deliver 50%+ cost savings for supported model architectures. Teams that default to GPU without evaluating Neuron compatibility leave significant savings on the table. The Neuron SDK supports PyTorch and TensorFlow natively with minimal code changes.

• Hardcoding instance types and hyperparameters in pipeline definitions. Non-parameterized pipelines cannot be reused across environments (dev/staging/prod) and require code changes for every configuration adjustment. Use SageMaker Pipeline parameters for all configurable values.

• Manual model deployment without CI/CD. Manual deployments are error-prone, unauditable, and slow. Use SageMaker Pipelines or CodePipeline to automate the path from model registration to staging to production, with approval gates at each stage.

Additional Resources

Reference Files

For detailed configurations, CLI commands, and code examples, consult:

• `references/training-patterns.md` — Training job configurations (single-instance, distributed, Spot), hyperparameter tuning setup, checkpointing, SageMaker Processing examples, and distributed training strategies
• `references/inference-deployment.md` — Real-time endpoint configurations, serverless inference, batch transform, async inference, auto-scaling policies, multi-model endpoints, shadow testing, and Inference Recommender usage
• `references/pipeline-recipes.md` — SageMaker Pipeline definitions (Python SDK), Model Registry workflows, CI/CD integration with CodePipeline, MLflow experiment tracking setup, and monitoring configuration

Related Skills

• `bedrock` — Foundation model customization, fine-tuning, and Bedrock-native inference
• `eks` — Kubernetes cluster design for Kubeflow or self-hosted MLflow deployments
• `lambda` — Serverless compute for lightweight ML inference or pipeline triggers
• `step-functions` — Workflow orchestration for simple ML pipelines without SageMaker Pipelines
• `s3` — Data lake design, artifact storage, lifecycle policies for training data and model artifacts
• `iam` — Least-privilege roles for SageMaker execution, cross-account model deployment
• `observability` — CloudWatch dashboards, alarms, and logging for ML infrastructure
• `cost-check` — Detailed cost analysis, Savings Plans recommendations, and Spot vs On-Demand comparison
• `ec2` — Instance type selection for self-managed training clusters or custom inference servers

Output Format

When recommending an MLOps architecture, include:

Include estimated monthly cost range using the cost-check skill.

MLOps Inference Deployment Reference

Real-Time Endpoint

Basic Endpoint Deployment

from sagemaker.pytorch import PyTorchModel

model = PyTorchModel(
    model_data=f"s3://{bucket}/output/model.tar.gz",
    role=sagemaker_role,
    framework_version="2.1.0",
    py_version="py310",
    entry_point="inference.py",
    source_dir="src/",
)

predictor = model.deploy(
    initial_instance_count=1,
    instance_type="ml.g5.xlarge",
    endpoint_name="my-model-endpoint",
    wait=True,
)

# Invoke
response = predictor.predict({"inputs": "Hello world"})

Deploying with Inferentia2 (ml.inf2)

from sagemaker.pytorch import PyTorchModel

# Model must be compiled with Neuron SDK
model = PyTorchModel(
    model_data=f"s3://{bucket}/output/neuron-model.tar.gz",
    role=sagemaker_role,
    image_uri=sagemaker.image_uris.retrieve(
        framework="pytorch",
        region=region,
        version="2.1.0",
        instance_type="ml.inf2.xlarge",
    ),
    entry_point="inference_neuron.py",
    source_dir="src/",
)

predictor = model.deploy(
    initial_instance_count=1,
    instance_type="ml.inf2.xlarge",  # 1 Inferentia2 chip, cost-effective inference
    endpoint_name="my-inf2-endpoint",
)

Deploy from Model Registry

from sagemaker import ModelPackage

model_package_arn = (
    "arn:aws:sagemaker:us-east-1:123456789012:model-package/my-model-group/1"
)

model = ModelPackage(
    role=sagemaker_role,
    model_package_arn=model_package_arn,
    sagemaker_session=sagemaker_session,
)

predictor = model.deploy(
    initial_instance_count=1,
    instance_type="ml.g5.xlarge",
    endpoint_name="my-production-endpoint",
)

Auto-Scaling Configuration

Target Tracking on InvocationsPerInstance

import boto3

client = boto3.client("application-autoscaling")

# Register the endpoint as a scalable target
client.register_scalable_target(
    ServiceNamespace="sagemaker",
    ResourceId=f"endpoint/{endpoint_name}/variant/AllTraffic",
    ScalableDimension="sagemaker:variant:DesiredInstanceCount",
    MinCapacity=1,
    MaxCapacity=10,
)

# Target tracking scaling policy
client.put_scaling_policy(
    PolicyName="InvocationsPerInstanceScaling",
    ServiceNamespace="sagemaker",
    ResourceId=f"endpoint/{endpoint_name}/variant/AllTraffic",
    ScalableDimension="sagemaker:variant:DesiredInstanceCount",
    PolicyType="TargetTrackingScaling",
    TargetTrackingScalingPolicyConfiguration={
        "TargetValue": 750.0,  # Target invocations per instance per minute
        "PredefinedMetricSpecification": {
            "PredefinedMetricType": "SageMakerVariantInvocationsPerInstance",
        },
        "ScaleInCooldown": 300,   # 5 min cooldown before scaling in
        "ScaleOutCooldown": 60,   # 1 min cooldown before scaling out
    },
)

Step Scaling for More Control

# Step scaling — add 2 instances when invocations > 1000/min
client.put_scaling_policy(
    PolicyName="HighTrafficStepScaling",
    ServiceNamespace="sagemaker",
    ResourceId=f"endpoint/{endpoint_name}/variant/AllTraffic",
    ScalableDimension="sagemaker:variant:DesiredInstanceCount",
    PolicyType="StepScaling",
    StepScalingPolicyConfiguration={
        "AdjustmentType": "ChangeInCapacity",
        "StepAdjustments": [
            {
                "MetricIntervalLowerBound": 0,
                "MetricIntervalUpperBound": 500,
                "ScalingAdjustment": 1,
            },
            {
                "MetricIntervalLowerBound": 500,
                "ScalingAdjustment": 2,
            },
        ],
        "Cooldown": 120,
    },
)

Serverless Inference

Deployment

from sagemaker.serverless import ServerlessInferenceConfig

serverless_config = ServerlessInferenceConfig(
    memory_size_in_mb=4096,       # 1024, 2048, 3072, 4096, 5120, or 6144
    max_concurrency=10,            # Max concurrent invocations
    provisioned_concurrency=0,     # 0 = no provisioned (pure on-demand)
)

predictor = model.deploy(
    serverless_inference_config=serverless_config,
    endpoint_name="my-serverless-endpoint",
)

When Serverless is Cost-Effective

Comparison at 100 requests/day, 500ms avg inference time:

Real-time ml.m5.large (always on):
  $0.134/hour * 730 hours = ~$98/month

Serverless (4 GB memory):
  Compute: 100 req * 500ms * $0.00001667/ms = $0.83/month
  Request: 100 req * $0.0000002 = negligible
  Total: ~$1/month

Break-even: Serverless is cheaper until ~60,000 requests/day at 500ms latency

Cold Start Mitigation

• Provisioned concurrency: Pre-warms a specified number of instances. Eliminates cold start but adds baseline cost.
• Model optimization: Smaller model artifacts load faster. Quantize or distill models to reduce cold start time.
• Warm-up invocations: Schedule periodic invocations via EventBridge to keep instances warm (workaround, not recommended for production SLAs).

Batch Transform

Basic Batch Transform

transformer = model.transformer(
    instance_count=4,
    instance_type="ml.m5.4xlarge",
    output_path=f"s3://{bucket}/batch-output/",
    strategy="MultiRecord",        # Process multiple records per request
    max_payload=6,                  # Max payload in MB
    max_concurrent_transforms=4,   # Parallel requests per instance
    assemble_with="Line",          # How to assemble output
)

transformer.transform(
    data=f"s3://{bucket}/batch-input/",
    content_type="text/csv",
    split_type="Line",             # Split input by line
    wait=True,
)

Batch Transform for Large Datasets

# For very large datasets, increase parallelism
transformer = model.transformer(
    instance_count=10,
    instance_type="ml.g5.2xlarge",     # GPU for DL models
    output_path=f"s3://{bucket}/batch-output/",
    max_concurrent_transforms=8,
    max_payload=100,                    # Up to 100 MB per record
)

CLI: Start a Batch Transform Job

aws sagemaker create-transform-job \
    --transform-job-name "batch-$(date +%Y%m%d-%H%M%S)" \
    --model-name "my-model" \
    --transform-input '{
        "DataSource": {
            "S3DataSource": {
                "S3DataType": "S3Prefix",
                "S3Uri": "s3://my-bucket/batch-input/"
            }
        },
        "ContentType": "text/csv",
        "SplitType": "Line"
    }' \
    --transform-output '{
        "S3OutputPath": "s3://my-bucket/batch-output/",
        "AssembleWith": "Line"
    }' \
    --transform-resources '{
        "InstanceType": "ml.m5.4xlarge",
        "InstanceCount": 4
    }'

Async Inference

Deployment

from sagemaker.async_inference import AsyncInferenceConfig

async_config = AsyncInferenceConfig(
    output_path=f"s3://{bucket}/async-output/",
    failure_path=f"s3://{bucket}/async-failures/",
    max_concurrent_invocations_per_instance=4,
    notification_config={
        "SuccessTopic": success_sns_topic_arn,
        "ErrorTopic": error_sns_topic_arn,
    },
)

predictor = model.deploy(
    initial_instance_count=1,
    instance_type="ml.g5.2xlarge",
    async_inference_config=async_config,
    endpoint_name="my-async-endpoint",
)

Scale-to-Zero for Async Endpoints

# Async endpoints can scale to 0 instances when idle
client.register_scalable_target(
    ServiceNamespace="sagemaker",
    ResourceId=f"endpoint/{endpoint_name}/variant/AllTraffic",
    ScalableDimension="sagemaker:variant:DesiredInstanceCount",
    MinCapacity=0,   # Scale to zero
    MaxCapacity=5,
)

# Scale based on queue depth
client.put_scaling_policy(
    PolicyName="QueueBasedScaling",
    ServiceNamespace="sagemaker",
    ResourceId=f"endpoint/{endpoint_name}/variant/AllTraffic",
    ScalableDimension="sagemaker:variant:DesiredInstanceCount",
    PolicyType="TargetTrackingScaling",
    TargetTrackingScalingPolicyConfiguration={
        "TargetValue": 5.0,
        "CustomizedMetricSpecification": {
            "MetricName": "ApproximateBacklogSizePerInstance",
            "Namespace": "AWS/SageMaker",
            "Dimensions": [
                {"Name": "EndpointName", "Value": endpoint_name},
            ],
            "Statistic": "Average",
        },
        "ScaleInCooldown": 600,
        "ScaleOutCooldown": 60,
    },
)

Invoke Async Endpoint

import boto3
import json

runtime = boto3.client("sagemaker-runtime")

# Upload input to S3
s3 = boto3.client("s3")
s3.put_object(
    Bucket=bucket,
    Key="async-input/request-001.json",
    Body=json.dumps({"inputs": "Large input data here..."}),
)

# Invoke — returns immediately with output location
response = runtime.invoke_endpoint_async(
    EndpointName="my-async-endpoint",
    InputLocation=f"s3://{bucket}/async-input/request-001.json",
    ContentType="application/json",
)

output_location = response["OutputLocation"]
# Poll output_location or use SNS notification to know when result is ready

Multi-Model Endpoints (MME)

Deployment

from sagemaker.multidatamodel import MultiDataModel

mme = MultiDataModel(
    name="my-multi-model",
    model_data_prefix=f"s3://{bucket}/models/",  # Directory containing model.tar.gz files
    model=model,  # Base model for container config
    sagemaker_session=sagemaker_session,
)

predictor = mme.deploy(
    initial_instance_count=2,
    instance_type="ml.g5.xlarge",
    endpoint_name="my-mme-endpoint",
)

Invoke a Specific Model

# Specify which model to invoke via TargetModel
response = predictor.predict(
    data=payload,
    target_model="customer-123/model.tar.gz",  # Relative path under model_data_prefix
)

Adding/Removing Models Dynamically

# Add a new model — just upload to S3, MME loads on first request
mme.add_model(
    model_data_source=f"s3://{bucket}/new-models/customer-456/model.tar.gz",
    model_data_path="customer-456/model.tar.gz",
)

# List loaded models
models = mme.list_models()

Shadow Testing

Create a Shadow Variant

import boto3

sm = boto3.client("sagemaker")

# Create endpoint with production + shadow variant
sm.create_endpoint_config(
    EndpointConfigName="shadow-test-config",
    ProductionVariants=[
        {
            "VariantName": "production",
            "ModelName": "current-model",
            "InstanceType": "ml.g5.xlarge",
            "InitialInstanceCount": 2,
            "InitialVariantWeight": 1.0,
        },
    ],
    ShadowProductionVariants=[
        {
            "VariantName": "shadow",
            "ModelName": "candidate-model",
            "InstanceType": "ml.g5.xlarge",
            "InitialInstanceCount": 1,
            "SamplingPercentage": 100,  # % of production traffic to mirror
        },
    ],
)

sm.update_endpoint(
    EndpointName="my-production-endpoint",
    EndpointConfigName="shadow-test-config",
)

Compare Shadow Results

Shadow variant responses are logged to S3 via Data Capture. Compare production vs shadow predictions:

# Enable data capture on both variants
data_capture_config = {
    "EnableCapture": True,
    "InitialSamplingPercentage": 100,
    "DestinationS3Uri": f"s3://{bucket}/data-capture/",
    "CaptureOptions": [
        {"CaptureMode": "Input"},
        {"CaptureMode": "Output"},
    ],
}

After collecting sufficient data (recommend at least 1 week of production traffic), compare metrics:

• Prediction distribution differences
• Latency p50/p95/p99
• Error rates
• Business metric impact (if measurable)

Inference Recommender

Run a Benchmark

sm = boto3.client("sagemaker")

# Default job — tests a curated set of instance types
response = sm.create_inference_recommendations_job(
    JobName="my-model-benchmark",
    JobType="Default",  # or "Advanced" for custom configs
    RoleArn=sagemaker_role,
    InputConfig={
        "ModelPackageVersionArn": model_package_arn,
        "JobDurationInSeconds": 7200,
    },
)

# Check results
result = sm.describe_inference_recommendations_job(
    JobName="my-model-benchmark"
)

for rec in result["InferenceRecommendations"]:
    print(f"Instance: {rec['EndpointConfiguration']['InstanceType']}")
    print(f"  Cost/hour: ${rec['Metrics']['CostPerHour']}")
    print(f"  Cost/inference: ${rec['Metrics']['CostPerInference']}")
    print(f"  Latency p50: {rec['Metrics']['ModelLatency']}ms")
    print(f"  Max invocations: {rec['Metrics']['MaxInvocations']}/min")

Advanced Benchmark with Custom Traffic

response = sm.create_inference_recommendations_job(
    JobName="my-model-advanced-benchmark",
    JobType="Advanced",
    RoleArn=sagemaker_role,
    InputConfig={
        "ModelPackageVersionArn": model_package_arn,
        "JobDurationInSeconds": 7200,
        "EndpointConfigurations": [
            {"InstanceType": "ml.g5.xlarge"},
            {"InstanceType": "ml.g5.2xlarge"},
            {"InstanceType": "ml.inf2.xlarge"},
            {"InstanceType": "ml.c7g.2xlarge"},
        ],
        "TrafficPattern": {
            "TrafficType": "PHASES",
            "Phases": [
                {"InitialNumberOfUsers": 1, "SpawnRate": 1, "DurationInSeconds": 300},
                {"InitialNumberOfUsers": 10, "SpawnRate": 2, "DurationInSeconds": 300},
                {"InitialNumberOfUsers": 50, "SpawnRate": 5, "DurationInSeconds": 300},
            ],
        },
    },
)

CLI Commands

Endpoint Management

# Create endpoint
aws sagemaker create-endpoint \
    --endpoint-name "my-endpoint" \
    --endpoint-config-name "my-config"

# Describe endpoint
aws sagemaker describe-endpoint \
    --endpoint-name "my-endpoint" \
    --query '{Status: EndpointStatus, Instance: ProductionVariants[0].CurrentInstanceCount}'

# Update endpoint (zero-downtime via rolling update)
aws sagemaker update-endpoint \
    --endpoint-name "my-endpoint" \
    --endpoint-config-name "my-new-config"

# Delete endpoint
aws sagemaker delete-endpoint \
    --endpoint-name "my-endpoint"

# List endpoints
aws sagemaker list-endpoints \
    --sort-by CreationTime \
    --sort-order Descending \
    --max-results 10

Invoke Endpoint

# Real-time invocation
aws sagemaker-runtime invoke-endpoint \
    --endpoint-name "my-endpoint" \
    --content-type "application/json" \
    --body '{"inputs": "test input"}' \
    output.json

# Check response
cat output.json

Endpoint Metrics

# Get invocation metrics for the last hour
aws cloudwatch get-metric-statistics \
    --namespace "AWS/SageMaker" \
    --metric-name "Invocations" \
    --dimensions Name=EndpointName,Value=my-endpoint Name=VariantName,Value=AllTraffic \
    --start-time "$(date -u -v-1H +%Y-%m-%dT%H:%M:%S)" \
    --end-time "$(date -u +%Y-%m-%dT%H:%M:%S)" \
    --period 300 \
    --statistics Sum

# Get model latency p99
aws cloudwatch get-metric-statistics \
    --namespace "AWS/SageMaker" \
    --metric-name "ModelLatency" \
    --dimensions Name=EndpointName,Value=my-endpoint Name=VariantName,Value=AllTraffic \
    --start-time "$(date -u -v-1H +%Y-%m-%dT%H:%M:%S)" \
    --end-time "$(date -u +%Y-%m-%dT%H:%M:%S)" \
    --period 300 \
    --statistics p99

MLOps Pipeline Recipes Reference

SageMaker Pipeline — Full Training Pipeline

End-to-End Pipeline Definition

import sagemaker
from sagemaker.workflow.pipeline import Pipeline
from sagemaker.workflow.parameters import ParameterString, ParameterFloat, ParameterInteger
from sagemaker.workflow.steps import ProcessingStep, TrainingStep, TransformStep
from sagemaker.workflow.step_collections import RegisterModel
from sagemaker.workflow.conditions import ConditionGreaterThanOrEqualTo
from sagemaker.workflow.condition_step import ConditionStep
from sagemaker.workflow.functions import JsonGet
from sagemaker.workflow.properties import PropertyFile
from sagemaker.sklearn.processing import SKLearnProcessor
from sagemaker.pytorch import PyTorch
from sagemaker.inputs import TrainingInput

# ── Pipeline Parameters (parameterize everything for reuse across envs) ──
input_data = ParameterString(name="InputData", default_value=f"s3://{bucket}/raw-data/")
instance_type_training = ParameterString(name="TrainingInstanceType", default_value="ml.g5.2xlarge")
instance_type_processing = ParameterString(name="ProcessingInstanceType", default_value="ml.m5.xlarge")
accuracy_threshold = ParameterFloat(name="AccuracyThreshold", default_value=0.85)
epochs = ParameterInteger(name="Epochs", default_value=10)
model_package_group = ParameterString(name="ModelPackageGroup", default_value="my-model-group")

# ── Step 1: Data Processing ──
sklearn_processor = SKLearnProcessor(
    framework_version="1.2-1",
    role=sagemaker_role,
    instance_type=instance_type_processing,
    instance_count=1,
    sagemaker_session=pipeline_session,
)

processing_step = ProcessingStep(
    name="PreprocessData",
    processor=sklearn_processor,
    code="scripts/preprocess.py",
    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"
        ),
        sagemaker.processing.ProcessingOutput(
            output_name="validation", source="/opt/ml/processing/output/validation"
        ),
        sagemaker.processing.ProcessingOutput(
            output_name="test", source="/opt/ml/processing/output/test"
        ),
    ],
    cache_config=CacheConfig(enable_caching=True, expire_after="P30D"),
)

# ── Step 2: Model Training ──
estimator = PyTorch(
    entry_point="train.py",
    source_dir="src/",
    role=sagemaker_role,
    instance_count=1,
    instance_type=instance_type_training,
    framework_version="2.1.0",
    py_version="py310",
    use_spot_instances=True,
    max_wait=7200,
    max_run=3600,
    checkpoint_s3_uri=f"s3://{bucket}/pipeline-checkpoints/",
    hyperparameters={
        "epochs": epochs,
        "batch-size": 64,
        "learning-rate": 0.001,
    },
    sagemaker_session=pipeline_session,
)

training_step = TrainingStep(
    name="TrainModel",
    estimator=estimator,
    inputs={
        "train": TrainingInput(
            s3_data=processing_step.properties.ProcessingOutputConfig.Outputs["train"].S3Output.S3Uri
        ),
        "validation": TrainingInput(
            s3_data=processing_step.properties.ProcessingOutputConfig.Outputs["validation"].S3Output.S3Uri
        ),
    },
    cache_config=CacheConfig(enable_caching=True, expire_after="P7D"),
)

# ── Step 3: Model Evaluation ──
evaluation_report = PropertyFile(
    name="EvaluationReport",
    output_name="evaluation",
    path="evaluation.json",
)

evaluation_step = ProcessingStep(
    name="EvaluateModel",
    processor=sklearn_processor,
    code="scripts/evaluate.py",
    inputs=[
        sagemaker.processing.ProcessingInput(
            source=training_step.properties.ModelArtifacts.S3ModelArtifacts,
            destination="/opt/ml/processing/model",
        ),
        sagemaker.processing.ProcessingInput(
            source=processing_step.properties.ProcessingOutputConfig.Outputs["test"].S3Output.S3Uri,
            destination="/opt/ml/processing/test",
        ),
    ],
    outputs=[
        sagemaker.processing.ProcessingOutput(
            output_name="evaluation",
            source="/opt/ml/processing/evaluation",
        ),
    ],
    property_files=[evaluation_report],
)

# ── Step 4: Conditional Registration ──
register_step = RegisterModel(
    name="RegisterModel",
    estimator=estimator,
    model_data=training_step.properties.ModelArtifacts.S3ModelArtifacts,
    content_types=["application/json"],
    response_types=["application/json"],
    inference_instances=["ml.g5.xlarge", "ml.inf2.xlarge", "ml.c7g.xlarge"],
    transform_instances=["ml.m5.xlarge"],
    model_package_group_name=model_package_group,
    approval_status="PendingManualApproval",
    model_metrics={
        "ModelQuality": {
            "Statistics": {
                "ContentType": "application/json",
                "S3Uri": f"s3://{bucket}/evaluation/statistics.json",
            }
        }
    },
)

# Quality gate: only register if accuracy exceeds threshold
condition = ConditionGreaterThanOrEqualTo(
    left=JsonGet(
        step_name=evaluation_step.name,
        property_file=evaluation_report,
        json_path="metrics.accuracy.value",
    ),
    right=accuracy_threshold,
)

condition_step = ConditionStep(
    name="CheckAccuracy",
    conditions=[condition],
    if_steps=[register_step],
    else_steps=[],  # Pipeline ends without registration if accuracy is too low
)

# ── Assemble Pipeline ──
pipeline = Pipeline(
    name="my-ml-pipeline",
    parameters=[
        input_data,
        instance_type_training,
        instance_type_processing,
        accuracy_threshold,
        epochs,
        model_package_group,
    ],
    steps=[processing_step, training_step, evaluation_step, condition_step],
    sagemaker_session=pipeline_session,
)

# Create or update the pipeline
pipeline.upsert(role_arn=sagemaker_role)

# Execute the pipeline
execution = pipeline.start(
    parameters={
        "InputData": f"s3://{bucket}/new-data/",
        "Epochs": 20,
        "AccuracyThreshold": 0.90,
    }
)

Evaluation Script (scripts/evaluate.py)

import json
import os
import tarfile
import torch
from sklearn.metrics import accuracy_score, precision_score, recall_score, f1_score

if __name__ == "__main__":
    # Load model
    model_path = "/opt/ml/processing/model/model.tar.gz"
    with tarfile.open(model_path) as tar:
        tar.extractall(path="/opt/ml/processing/model/extracted")

    model = torch.load("/opt/ml/processing/model/extracted/model.pth")
    model.eval()

    # Load test data
    test_data = load_test_data("/opt/ml/processing/test/")

    # Run predictions
    predictions = []
    labels = []
    with torch.no_grad():
        for batch in test_data:
            outputs = model(batch["inputs"])
            predictions.extend(outputs.argmax(dim=1).tolist())
            labels.extend(batch["labels"].tolist())

    # Calculate metrics
    accuracy = accuracy_score(labels, predictions)
    precision = precision_score(labels, predictions, average="weighted")
    recall = recall_score(labels, predictions, average="weighted")
    f1 = f1_score(labels, predictions, average="weighted")

    # Write evaluation report
    report = {
        "metrics": {
            "accuracy": {"value": accuracy},
            "precision": {"value": precision},
            "recall": {"value": recall},
            "f1": {"value": f1},
        }
    }

    output_dir = "/opt/ml/processing/evaluation"
    os.makedirs(output_dir, exist_ok=True)
    with open(os.path.join(output_dir, "evaluation.json"), "w") as f:
        json.dump(report, f)

    print(f"Accuracy: {accuracy:.4f}, Precision: {precision:.4f}, "
          f"Recall: {recall:.4f}, F1: {f1:.4f}")

Model Registry Workflows

Register a Model Manually

from sagemaker import ModelPackage

model_package = sm_client.create_model_package(
    ModelPackageGroupName="my-model-group",
    ModelPackageDescription="v2.1 — improved accuracy on edge cases",
    InferenceSpecification={
        "Containers": [
            {
                "Image": container_image_uri,
                "ModelDataUrl": f"s3://{bucket}/models/model-v2.1.tar.gz",
            }
        ],
        "SupportedContentTypes": ["application/json"],
        "SupportedResponseMIMETypes": ["application/json"],
        "SupportedRealtimeInferenceInstanceTypes": [
            "ml.g5.xlarge", "ml.inf2.xlarge",
        ],
        "SupportedTransformInstanceTypes": ["ml.m5.xlarge"],
    },
    ModelApprovalStatus="PendingManualApproval",
    ModelMetrics={
        "ModelQuality": {
            "Statistics": {
                "ContentType": "application/json",
                "S3Uri": f"s3://{bucket}/evaluation/v2.1/metrics.json",
            }
        }
    },
)

Approve a Model

sm_client.update_model_package(
    ModelPackageArn=model_package_arn,
    ModelApprovalStatus="Approved",
    ApprovalDescription="Reviewed by ML team. Accuracy 94.2% on holdout set.",
)

Automated Approval via EventBridge

# EventBridge rule: trigger deployment when a model is approved
import json

rule = {
    "source": ["aws.sagemaker"],
    "detail-type": ["SageMaker Model Package State Change"],
    "detail": {
        "ModelPackageGroupName": ["my-model-group"],
        "ModelApprovalStatus": ["Approved"],
    },
}

# Target: CodePipeline or Lambda that deploys the approved model
events_client.put_rule(
    Name="model-approved-trigger",
    EventPattern=json.dumps(rule),
    State="ENABLED",
)

events_client.put_targets(
    Rule="model-approved-trigger",
    Targets=[
        {
            "Id": "deploy-pipeline",
            "Arn": codepipeline_arn,
            "RoleArn": eventbridge_role_arn,
        }
    ],
)

Cross-Account Model Deployment

# In the model-producing account: grant cross-account access
sm_client.put_model_package_group_policy(
    ModelPackageGroupName="my-model-group",
    ResourcePolicy=json.dumps({
        "Version": "2012-10-17",
        "Statement": [
            {
                "Sid": "AllowProductionAccountAccess",
                "Effect": "Allow",
                "Principal": {
                    "AWS": "arn:aws:iam::PRODUCTION_ACCOUNT_ID:root"
                },
                "Action": [
                    "sagemaker:DescribeModelPackage",
                    "sagemaker:DescribeModelPackageGroup",
                    "sagemaker:ListModelPackages",
                ],
                "Resource": "*",
            }
        ],
    }),
)

# In the production account: deploy the model using its ARN
model = ModelPackage(
    role=production_role,
    model_package_arn=f"arn:aws:sagemaker:us-east-1:MODEL_ACCOUNT_ID:model-package/my-model-group/3",
)
predictor = model.deploy(
    instance_type="ml.g5.xlarge",
    initial_instance_count=2,
)

CI/CD Integration

CodePipeline + SageMaker Pipeline

# buildspec.yml for CodeBuild stage that triggers SageMaker Pipeline
version: 0.2

phases:
  install:
    runtime-versions:
      python: 3.11
    commands:
      - pip install sagemaker boto3

  build:
    commands:
      - echo "Starting SageMaker Pipeline execution"
      - python scripts/start_pipeline.py

  post_build:
    commands:
      - echo "Pipeline execution started"
      - python scripts/wait_for_pipeline.py

Pipeline Trigger Script (scripts/start_pipeline.py)

import boto3
import json
import os

sm = boto3.client("sagemaker")

pipeline_name = os.environ.get("PIPELINE_NAME", "my-ml-pipeline")
commit_id = os.environ.get("CODEBUILD_RESOLVED_SOURCE_VERSION", "unknown")

# Start pipeline execution with parameters
response = sm.start_pipeline_execution(
    PipelineName=pipeline_name,
    PipelineExecutionDisplayName=f"ci-{commit_id[:8]}",
    PipelineParameters=[
        {"Name": "InputData", "Value": f"s3://{os.environ['DATA_BUCKET']}/latest/"},
        {"Name": "Epochs", "Value": "20"},
    ],
    PipelineExecutionDescription=f"Triggered by commit {commit_id}",
)

execution_arn = response["PipelineExecutionArn"]
print(f"Pipeline execution started: {execution_arn}")

# Save ARN for the wait step
with open("pipeline_execution_arn.txt", "w") as f:
    f.write(execution_arn)

GitHub Actions Integration

# .github/workflows/ml-pipeline.yml
name: ML Pipeline

on:
  push:
    branches: [main]
    paths:
      - 'src/training/**'
      - 'scripts/**'
      - 'configs/**'

jobs:
  trigger-pipeline:
    runs-on: ubuntu-latest
    permissions:
      id-token: write
      contents: read

    steps:
      - uses: actions/checkout@v4

      - uses: aws-actions/configure-aws-credentials@v4
        with:
          role-to-assume: ${{ secrets.SAGEMAKER_ROLE_ARN }}
          aws-region: us-east-1

      - uses: actions/setup-python@v5
        with:
          python-version: '3.11'

      - name: Install dependencies
        run: pip install sagemaker boto3

      - name: Update and start pipeline
        run: |
          python scripts/upsert_pipeline.py
          python scripts/start_pipeline.py
        env:
          DATA_BUCKET: ${{ secrets.DATA_BUCKET }}
          PIPELINE_NAME: my-ml-pipeline

      - name: Wait for pipeline completion
        run: python scripts/wait_for_pipeline.py
        timeout-minutes: 120

MLflow Experiment Tracking

Managed MLflow on SageMaker Setup

import mlflow
import sagemaker

# Get the MLflow tracking URI from SageMaker
tracking_server_arn = "arn:aws:sagemaker:us-east-1:123456789012:mlflow-tracking-server/my-server"
tracking_uri = sagemaker.session.Session().sagemaker_client.describe_mlflow_tracking_server(
    TrackingServerName="my-server"
)["TrackingServerUrl"]

mlflow.set_tracking_uri(tracking_uri)

Experiment Tracking in Training Script

import mlflow
import mlflow.pytorch

# Set experiment (creates if not exists)
mlflow.set_experiment("my-classification-project")

with mlflow.start_run(run_name="pytorch-v2.1") as run:
    # Log parameters
    mlflow.log_params({
        "learning_rate": 0.001,
        "batch_size": 64,
        "epochs": 20,
        "optimizer": "AdamW",
        "model_architecture": "resnet50",
        "instance_type": "ml.g5.2xlarge",
    })

    # Training loop
    for epoch in range(epochs):
        train_loss = train_one_epoch(model, optimizer, train_loader)
        val_loss, val_accuracy = evaluate(model, val_loader)

        # Log metrics per epoch
        mlflow.log_metrics({
            "train_loss": train_loss,
            "val_loss": val_loss,
            "val_accuracy": val_accuracy,
        }, step=epoch)

    # Log the final model
    mlflow.pytorch.log_model(
        model,
        "model",
        registered_model_name="my-classifier",  # Auto-registers in SageMaker Model Registry
    )

    # Log artifacts
    mlflow.log_artifact("confusion_matrix.png")
    mlflow.log_artifact("classification_report.json")

    print(f"Run ID: {run.info.run_id}")

Compare Experiments

import mlflow

# Search runs across experiments
runs = mlflow.search_runs(
    experiment_names=["my-classification-project"],
    filter_string="metrics.val_accuracy > 0.85",
    order_by=["metrics.val_accuracy DESC"],
    max_results=10,
)

print(runs[["run_id", "params.learning_rate", "params.batch_size",
            "metrics.val_accuracy", "metrics.val_loss"]])

Deploy MLflow Model to SageMaker

import mlflow.sagemaker

# Deploy directly from MLflow model registry
mlflow.sagemaker.deploy(
    model_uri="models:/my-classifier/Production",
    endpoint_name="my-mlflow-endpoint",
    region_name="us-east-1",
    instance_type="ml.g5.xlarge",
    instance_count=1,
    role=sagemaker_role,
)

Model Monitoring Configuration

Data Quality Monitor

from sagemaker.model_monitor import DefaultModelMonitor
from sagemaker.model_monitor.dataset_format import DatasetFormat

# Create baseline from training data
monitor = DefaultModelMonitor(
    role=sagemaker_role,
    instance_count=1,
    instance_type="ml.m5.xlarge",
    volume_size_in_gb=20,
    max_runtime_in_seconds=3600,
)

monitor.suggest_baseline(
    baseline_dataset=f"s3://{bucket}/data/train/train.csv",
    dataset_format=DatasetFormat.csv(header=True),
    output_s3_uri=f"s3://{bucket}/monitoring/baseline/",
)

# Schedule monitoring
monitor.create_monitoring_schedule(
    monitor_schedule_name="data-quality-monitor",
    endpoint_input=endpoint_name,
    output_s3_uri=f"s3://{bucket}/monitoring/data-quality-reports/",
    statistics=monitor.baseline_statistics(),
    constraints=monitor.suggested_constraints(),
    schedule_cron_expression="cron(0 * ? * * *)",  # Hourly
)

Model Quality Monitor

from sagemaker.model_monitor import ModelQualityMonitor

model_monitor = ModelQualityMonitor(
    role=sagemaker_role,
    instance_count=1,
    instance_type="ml.m5.xlarge",
    volume_size_in_gb=20,
    max_runtime_in_seconds=1800,
    sagemaker_session=sagemaker_session,
)

# Create baseline
model_monitor.suggest_baseline(
    problem_type="BinaryClassification",
    baseline_dataset=f"s3://{bucket}/baseline/predictions-with-labels.csv",
    dataset_format=DatasetFormat.csv(header=True),
    output_s3_uri=f"s3://{bucket}/monitoring/model-quality-baseline/",
    ground_truth_input=f"s3://{bucket}/ground-truth/",
)

# Schedule
model_monitor.create_monitoring_schedule(
    monitor_schedule_name="model-quality-monitor",
    endpoint_input=endpoint_name,
    output_s3_uri=f"s3://{bucket}/monitoring/model-quality-reports/",
    problem_type="BinaryClassification",
    ground_truth_input=f"s3://{bucket}/ground-truth/",
    constraints=model_monitor.suggested_constraints(),
    schedule_cron_expression="cron(0 0 ? * * *)",  # Daily
)

CloudWatch Alarms for Monitoring Violations

import boto3

cloudwatch = boto3.client("cloudwatch")

# Alarm on data quality violations
cloudwatch.put_metric_alarm(
    AlarmName="mlops-data-quality-violation",
    MetricName="data_quality_violations",
    Namespace="aws/sagemaker/Endpoints/data-metrics",
    Statistic="Maximum",
    Period=3600,
    EvaluationPeriods=1,
    Threshold=0,
    ComparisonOperator="GreaterThanThreshold",
    AlarmActions=[sns_topic_arn],
    AlarmDescription="Data quality violation detected — feature distribution drift",
    Dimensions=[
        {"Name": "Endpoint", "Value": endpoint_name},
        {"Name": "MonitoringSchedule", "Value": "data-quality-monitor"},
    ],
)

Automated Retraining on Drift Detection

# EventBridge rule: trigger pipeline when monitoring detects violations
rule = {
    "source": ["aws.sagemaker"],
    "detail-type": ["SageMaker Model Monitor Alert"],
    "detail": {
        "MonitoringScheduleName": ["data-quality-monitor"],
    },
}

# Target: Lambda that starts the SageMaker Pipeline
events_client.put_rule(
    Name="drift-detected-retrain",
    EventPattern=json.dumps(rule),
    State="ENABLED",
)

events_client.put_targets(
    Rule="drift-detected-retrain",
    Targets=[
        {
            "Id": "retrain-trigger",
            "Arn": retrain_lambda_arn,
            "RoleArn": eventbridge_role_arn,
        }
    ],
)

CLI Commands

Pipeline Management

# List pipelines
aws sagemaker list-pipelines \
    --sort-by CreationTime \
    --sort-order Descending \
    --max-results 10

# Describe a pipeline
aws sagemaker describe-pipeline \
    --pipeline-name "my-ml-pipeline"

# Start pipeline execution
aws sagemaker start-pipeline-execution \
    --pipeline-name "my-ml-pipeline" \
    --pipeline-parameters '[
        {"Name": "InputData", "Value": "s3://my-bucket/new-data/"},
        {"Name": "Epochs", "Value": "20"}
    ]'

# List executions
aws sagemaker list-pipeline-executions \
    --pipeline-name "my-ml-pipeline" \
    --sort-by CreationTime \
    --sort-order Descending \
    --max-results 5

# Describe execution
aws sagemaker describe-pipeline-execution \
    --pipeline-execution-arn "arn:aws:sagemaker:us-east-1:123456789012:pipeline/my-ml-pipeline/execution/abc123"

# List steps in an execution
aws sagemaker list-pipeline-execution-steps \
    --pipeline-execution-arn "arn:aws:sagemaker:us-east-1:123456789012:pipeline/my-ml-pipeline/execution/abc123"

# Stop a running pipeline
aws sagemaker stop-pipeline-execution \
    --pipeline-execution-arn "arn:aws:sagemaker:us-east-1:123456789012:pipeline/my-ml-pipeline/execution/abc123"

Model Registry

# List model package groups
aws sagemaker list-model-package-groups \
    --sort-by CreationTime \
    --sort-order Descending

# List model versions in a group
aws sagemaker list-model-packages \
    --model-package-group-name "my-model-group" \
    --sort-by CreationTime \
    --sort-order Descending

# Describe a model version
aws sagemaker describe-model-package \
    --model-package-name "arn:aws:sagemaker:us-east-1:123456789012:model-package/my-model-group/3"

# Approve a model
aws sagemaker update-model-package \
    --model-package-arn "arn:aws:sagemaker:us-east-1:123456789012:model-package/my-model-group/3" \
    --model-approval-status "Approved" \
    --approval-description "Approved after staging validation"

Monitoring

# List monitoring schedules
aws sagemaker list-monitoring-schedules \
    --endpoint-name "my-endpoint" \
    --sort-by CreationTime

# Describe monitoring schedule
aws sagemaker describe-monitoring-schedule \
    --monitoring-schedule-name "data-quality-monitor"

# List monitoring executions
aws sagemaker list-monitoring-executions \
    --monitoring-schedule-name "data-quality-monitor" \
    --sort-by CreationTime \
    --sort-order Descending \
    --max-results 5

# Check latest violation report
aws s3 cp s3://my-bucket/monitoring/data-quality-reports/latest/constraint_violations.json - | jq .

MLOps Training Patterns Reference

Single-Instance Training Job

Basic Training Job (PyTorch)

from sagemaker.pytorch import PyTorch

estimator = PyTorch(
    entry_point="train.py",
    source_dir="src/",
    role=sagemaker_role,
    instance_count=1,
    instance_type="ml.g5.2xlarge",
    framework_version="2.1.0",
    py_version="py310",
    # Spot training — 60-90% savings
    use_spot_instances=True,
    max_wait=7200,       # 2x expected training time
    max_run=3600,        # max training time in seconds
    # Checkpointing for Spot resilience
    checkpoint_s3_uri=f"s3://{bucket}/checkpoints/{job_name}",
    checkpoint_local_path="/opt/ml/checkpoints",
    # Environment
    hyperparameters={
        "epochs": 10,
        "batch-size": 64,
        "learning-rate": 0.001,
    },
    tags=[{"Key": "project", "Value": "my-ml-project"}],
)

estimator.fit({
    "train": f"s3://{bucket}/data/train/",
    "validation": f"s3://{bucket}/data/validation/",
})

Training with Trainium (ml.trn1)

from sagemaker.pytorch import PyTorch

estimator = PyTorch(
    entry_point="train_neuron.py",
    source_dir="src/",
    role=sagemaker_role,
    instance_count=1,
    instance_type="ml.trn1.32xlarge",  # 16 Trainium chips, 512 GB accelerator memory
    framework_version="2.1.0",
    py_version="py310",
    # Neuron SDK is included in the SageMaker Trainium DLC
    image_uri=sagemaker.image_uris.retrieve(
        framework="pytorch",
        region=region,
        version="2.1.0",
        instance_type="ml.trn1.32xlarge",
    ),
    use_spot_instances=True,
    max_wait=14400,
    max_run=7200,
    checkpoint_s3_uri=f"s3://{bucket}/checkpoints/{job_name}",
    hyperparameters={
        "epochs": 10,
        "batch-size": 128,
    },
    distribution={
        "torch_distributed": {
            "enabled": True,
        }
    },
)

Classical ML Training (XGBoost)

from sagemaker.xgboost import XGBoost

estimator = XGBoost(
    entry_point="train.py",
    role=sagemaker_role,
    instance_count=1,
    instance_type="ml.m5.2xlarge",  # CPU only — no GPU needed for tree models
    framework_version="1.7-1",
    use_spot_instances=True,
    max_wait=3600,
    max_run=1800,
    hyperparameters={
        "max_depth": 6,
        "eta": 0.3,
        "num_round": 200,
        "objective": "binary:logistic",
        "eval_metric": "auc",
    },
)

Distributed Training

Data Parallel Training (SageMaker Distributed Data Parallelism)

Use when the model fits in one GPU but training is slow due to dataset size.

from sagemaker.pytorch import PyTorch

estimator = PyTorch(
    entry_point="train_ddp.py",
    source_dir="src/",
    role=sagemaker_role,
    instance_count=4,              # 4 nodes
    instance_type="ml.p4d.24xlarge",  # 8x A100 per node = 32 GPUs total
    framework_version="2.1.0",
    py_version="py310",
    use_spot_instances=True,
    max_wait=14400,
    max_run=7200,
    checkpoint_s3_uri=f"s3://{bucket}/checkpoints/{job_name}",
    distribution={
        "smdistributed": {
            "dataparallel": {
                "enabled": True,
            }
        }
    },
    hyperparameters={
        "epochs": 20,
        "batch-size": 256,  # Global batch size = 256 * 32 GPUs
        "learning-rate": 0.001,
    },
)

Training script changes for SMDDP:

import torch
import smdistributed.dataparallel.torch.torch_smddp  # Initialize SMDDP

# Use PyTorch DDP as normal — SMDDP replaces the backend
torch.distributed.init_process_group(backend="smddp")

local_rank = int(os.environ["LOCAL_RANK"])
torch.cuda.set_device(local_rank)

model = MyModel().to(local_rank)
model = torch.nn.parallel.DistributedDataParallel(model, device_ids=[local_rank])

Model Parallel Training (SageMaker Model Parallelism)

Use when the model does not fit in a single GPU's memory.

from sagemaker.pytorch import PyTorch

estimator = PyTorch(
    entry_point="train_mp.py",
    source_dir="src/",
    role=sagemaker_role,
    instance_count=2,
    instance_type="ml.p5.48xlarge",  # 8x H100 per node, EFA enabled
    framework_version="2.1.0",
    py_version="py310",
    use_spot_instances=True,
    max_wait=28800,
    max_run=14400,
    checkpoint_s3_uri=f"s3://{bucket}/checkpoints/{job_name}",
    distribution={
        "smdistributed": {
            "modelparallel": {
                "enabled": True,
                "parameters": {
                    "tensor_parallel_degree": 8,
                    "pipeline_parallel_degree": 2,
                    "ddp": True,
                }
            }
        }
    },
)

PyTorch Native Distributed (torchrun)

Use when you want framework-native distributed training without SageMaker libraries.

estimator = PyTorch(
    entry_point="train.py",
    source_dir="src/",
    role=sagemaker_role,
    instance_count=2,
    instance_type="ml.g5.12xlarge",  # 4x A10G per node
    framework_version="2.1.0",
    py_version="py310",
    use_spot_instances=True,
    max_wait=7200,
    max_run=3600,
    distribution={
        "torch_distributed": {
            "enabled": True,
        }
    },
)

Managed Spot Training

Checkpointing Setup

Checkpointing is mandatory for Spot training. Without it, a Spot interruption restarts training from epoch 0.

In the training script:

import os
import torch

CHECKPOINT_DIR = "/opt/ml/checkpoints"

def save_checkpoint(model, optimizer, epoch, loss):
    """Save checkpoint to local path — SageMaker syncs to S3 automatically."""
    checkpoint = {
        "epoch": epoch,
        "model_state_dict": model.state_dict(),
        "optimizer_state_dict": optimizer.state_dict(),
        "loss": loss,
    }
    path = os.path.join(CHECKPOINT_DIR, f"checkpoint-{epoch}.pt")
    torch.save(checkpoint, path)

def load_latest_checkpoint(model, optimizer):
    """Resume from latest checkpoint if one exists (Spot restart)."""
    if not os.path.exists(CHECKPOINT_DIR):
        return 0
    checkpoints = sorted(
        [f for f in os.listdir(CHECKPOINT_DIR) if f.startswith("checkpoint-")],
        key=lambda x: int(x.split("-")[1].split(".")[0]),
    )
    if not checkpoints:
        return 0
    latest = os.path.join(CHECKPOINT_DIR, checkpoints[-1])
    checkpoint = torch.load(latest)
    model.load_state_dict(checkpoint["model_state_dict"])
    optimizer.load_state_dict(checkpoint["optimizer_state_dict"])
    return checkpoint["epoch"] + 1

# In training loop
start_epoch = load_latest_checkpoint(model, optimizer)
for epoch in range(start_epoch, total_epochs):
    train_one_epoch(model, optimizer, train_loader)
    save_checkpoint(model, optimizer, epoch, loss)

Spot Savings Calculation

On-Demand ml.p4d.24xlarge: ~$32.77/hour
Spot ml.p4d.24xlarge: ~$9.83/hour (typical 70% savings)

10-hour training job:
  On-Demand: $327.70
  Spot: $98.30
  Savings: $229.40 per job

With checkpointing, even if the job is interrupted twice (adding 30 min overhead each time), total cost is still ~$108 — 67% less than On-Demand.

Hyperparameter Tuning

Bayesian Optimization (Default)

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

hyperparameter_ranges = {
    "learning-rate": ContinuousParameter(1e-5, 1e-2, scaling_type="Logarithmic"),
    "batch-size": CategoricalParameter([32, 64, 128, 256]),
    "weight-decay": ContinuousParameter(1e-6, 1e-2, scaling_type="Logarithmic"),
    "num-layers": IntegerParameter(2, 8),
}

tuner = HyperparameterTuner(
    estimator=estimator,
    objective_metric_name="validation:accuracy",
    objective_type="Maximize",
    hyperparameter_ranges=hyperparameter_ranges,
    max_jobs=50,          # Total trials
    max_parallel_jobs=5,  # Parallel trials (Bayesian benefits from sequential info)
    strategy="Bayesian",  # Default and recommended
    early_stopping_type="Auto",  # Stop poor trials early
)

tuner.fit({
    "train": train_input,
    "validation": validation_input,
})

Hyperband Strategy

Use for faster results on a budget. Automatically allocates more resources to promising configurations.

from sagemaker.tuner import HyperparameterTuner

tuner = HyperparameterTuner(
    estimator=estimator,
    objective_metric_name="validation:loss",
    objective_type="Minimize",
    hyperparameter_ranges=hyperparameter_ranges,
    strategy="Hyperband",
    max_jobs=100,
    max_parallel_jobs=10,
    strategy_config={
        "HyperbandStrategyConfig": {
            "MinResource": 1,   # Min epochs before early stop
            "MaxResource": 50,  # Max epochs for best configs
        }
    },
)

SageMaker Processing Jobs

Data Preparation with sklearn

from sagemaker.processing import ScriptProcessor
from sagemaker.sklearn import SKLearnProcessor

processor = SKLearnProcessor(
    framework_version="1.2-1",
    role=sagemaker_role,
    instance_type="ml.m5.xlarge",
    instance_count=1,
)

processor.run(
    code="scripts/preprocess.py",
    inputs=[
        ProcessingInput(
            source=f"s3://{bucket}/raw-data/",
            destination="/opt/ml/processing/input",
        )
    ],
    outputs=[
        ProcessingOutput(
            output_name="train",
            source="/opt/ml/processing/output/train",
            destination=f"s3://{bucket}/processed/train/",
        ),
        ProcessingOutput(
            output_name="validation",
            source="/opt/ml/processing/output/validation",
            destination=f"s3://{bucket}/processed/validation/",
        ),
        ProcessingOutput(
            output_name="test",
            source="/opt/ml/processing/output/test",
            destination=f"s3://{bucket}/processed/test/",
        ),
    ],
)

Spark Processing for Large Datasets

from sagemaker.spark.processing import PySparkProcessor

spark_processor = PySparkProcessor(
    base_job_name="spark-preprocessing",
    framework_version="3.3",
    role=sagemaker_role,
    instance_count=4,
    instance_type="ml.m5.4xlarge",
    max_runtime_in_seconds=7200,
)

spark_processor.run(
    submit_app="scripts/spark_preprocess.py",
    arguments=[
        "--input-path", f"s3://{bucket}/raw-data/",
        "--output-path", f"s3://{bucket}/processed/",
    ],
    spark_event_logs_s3_uri=f"s3://{bucket}/spark-logs/",
)

CLI Commands

Launch a Training Job

aws sagemaker create-training-job \
    --training-job-name "my-training-$(date +%Y%m%d-%H%M%S)" \
    --algorithm-specification \
        TrainingImage="763104351884.dkr.ecr.us-east-1.amazonaws.com/pytorch-training:2.1.0-gpu-py310-cu118-ubuntu20.04-sagemaker" \
        TrainingInputMode=File \
    --role-arn "$SAGEMAKER_ROLE_ARN" \
    --resource-config \
        InstanceType=ml.g5.2xlarge,InstanceCount=1,VolumeSizeInGB=50 \
    --input-data-config '[{
        "ChannelName": "train",
        "DataSource": {
            "S3DataSource": {
                "S3DataType": "S3Prefix",
                "S3Uri": "s3://my-bucket/data/train/"
            }
        }
    }]' \
    --output-data-config S3OutputPath="s3://my-bucket/output/" \
    --stopping-condition MaxRuntimeInSeconds=3600 \
    --enable-managed-spot-training \
    --checkpoint-config S3Uri="s3://my-bucket/checkpoints/"

Monitor a Training Job

# Watch training job status
aws sagemaker describe-training-job \
    --training-job-name "my-training-job" \
    --query '{Status: TrainingJobStatus, Secondary: SecondaryStatus, Metrics: FinalMetricDataList}'

# Stream training logs
aws logs tail /aws/sagemaker/TrainingJobs --follow \
    --log-stream-name-prefix "my-training-job"

# List recent training jobs
aws sagemaker list-training-jobs \
    --sort-by CreationTime \
    --sort-order Descending \
    --max-results 10 \
    --query 'TrainingJobSummaries[].{Name:TrainingJobName,Status:TrainingJobStatus,Instance:ResourceConfig.InstanceType}'

Hyperparameter Tuning Job Status

aws sagemaker describe-hyper-parameter-tuning-job \
    --hyper-parameter-tuning-job-name "my-tuning-job" \
    --query '{
        Status: HyperParameterTuningJobStatus,
        BestTrainingJob: BestTrainingJob.{Name:TrainingJobName,Metric:FinalHyperParameterTuningJobObjectiveMetric},
        Completed: TrainingJobStatusCounters.Completed,
        InProgress: TrainingJobStatusCounters.InProgress
    }'