Tutorial Lengkap AWS SageMaker Model Monitor: Monitoring Model ML di Production

Amazon SageMaker Model Monitor secara otomatis mendeteksi masalah kualitas data, degradasi kualitas model, bias drift, dan feature attribution drift pada model ML yang di-deploy ke production. Layanan ini membantu mempertahankan performa model seiring waktu.

Mengapa Model Monitor?

• Monitoring otomatis: Pengawasan model berkelanjutan
• Deteksi drift: Alert untuk data dan model quality drift
• Deteksi bias: Monitor metrik fairness
• Explainability: Tracking feature attribution
• Integrasi: Integrasi native dengan SageMaker

• Data Quality Monitor
• Model Quality Monitor
• Bias Drift Monitor
• Feature Attribution Drift Monitor

Prerequisites

pip install sagemaker boto3 pandas numpy

SageMaker SDK >= 2.0
python -c "import sagemaker; print(sagemaker.version)"

Quick Start

1. Setup

import boto3
import sagemaker
from sagemaker import getexecutionrole
from sagemaker.modelmonitor import (

    DefaultModelMonitor,
    DataCaptureConfig,
    CronExpressionGenerator
)

session = sagemaker.Session()
bucket = session.defaultbucket()
role = getexecutionrole()
region = session.botoregionname

Lokasi output monitor
monitoroutput = f"s3://{bucket}/model-monitor"

from sagemaker.model import Model
from sagemaker.predictor import Predictor

Buat model
model = Model(
    imageuri=xgboostimage,

    modeldata=modeldatauri,
    role=role
)

Konfigurasi data capture
datacaptureconfig = DataCaptureConfig(
    enablecapture=True,

    samplingpercentage=100,  # Capture semua request
    destinations3uri=f"s3://{bucket}/data-capture",
    captureoptions=["Input", "Output"],

    csvcontenttypes=["text/csv"],

    jsoncontenttypes=["application/json"]

)

Deploy dengan data capture
predictor = model.deploy(
    initialinstancecount=1,

    instancetype="ml.m5.large",
    endpointname="monitored-endpoint",

    datacaptureconfig=datacaptureconfig

)

print(f"Endpoint di-deploy: {predictor.endpointname}")

Data Quality Monitor

1. Buat Baseline

from sagemaker.modelmonitor import DefaultModelMonitor
from sagemaker.modelmonitor.datasetformat import DatasetFormat


Buat monitor
dataqualitymonitor = DefaultModelMonitor(

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

    volumesizeingb=20,
    maxruntimeinseconds=3600

)

Buat baseline dari data training
dataqualitymonitor.suggestbaseline(
    baselinedataset=f"s3://{bucket}/training-data/train.csv",

    datasetformat=DatasetFormat.csv(header=True),
    outputs3uri=f"{monitoroutput}/data-quality/baseline",

    wait=True
)

print("Baseline dibuat!")

import json

Dapatkan statistik baseline
baselinejob = dataqualitymonitor.latestbaseliningjob

statisticspath = f"{monitoroutput}/data-quality/baseline/statistics.json"
constraintspath = f"{monitoroutput}/data-quality/baseline/constraints.json"

Download dan lihat statistik
s3 = boto3.client("s3")

Parse S3 URI
def parses3uri(uri):
    parts = uri.replace("s3://", "").split("/", 1)
    return parts[0], parts[1]

bucketname, key = parses3uri(statisticspath)
response = s3.getobject(Bucket=bucketname, Key=key)
statistics = json.loads(response["Body"].read())

print("Statistik Baseline:")
for feature in statistics["features"]:
    print(f"  {feature['name']}: mean={feature.get('numericalstatistics', {}).get('mean', 'N/A')}")

from sagemaker.modelmonitor import CronExpressionGenerator

Buat jadwal monitoring
dataqualitymonitor.createmonitoringschedule(
    monitorschedulename="data-quality-schedule",
    endpointinput=predictor.endpointname,
    outputs3uri=f"{monitoroutput}/data-quality/reports",

    statistics=dataqualitymonitor.baselinestatistics(),
    constraints=dataqualitymonitor.suggestedconstraints(),

    schedulecronexpression=CronExpressionGenerator.hourly(),

    enablecloudwatchmetrics=True

)

print("Jadwal monitoring dibuat!")

from sagemaker.modelmonitor import ModelQualityMonitor

Buat model quality monitor
modelqualitymonitor = ModelQualityMonitor(
    role=role,
    instancecount=1,

    instancetype="ml.m5.xlarge",
    volumesizeingb=20,

    maxruntimeinseconds=3600,
    sagemakersession=session

)

Buat baseline dengan ground truth
modelqualitymonitor.suggestbaseline(
    baselinedataset=f"s3://{bucket}/ground-truth/baseline.csv",

    datasetformat=DatasetFormat.csv(header=True),
    outputs3uri=f"{monitoroutput}/model-quality/baseline",

    problemtype="BinaryClassification",
    inferenceattribute="prediction",

    groundtruthattribute="label",

    wait=True
)

import pandas as pd
from datetime import datetime

Format data ground truth
groundtruthdata = pd.DataFrame({

    "inferenceid": ["id-001", "id-002", "id-003"],
    "prediction": [1, 0, 1],
    "label": [1, 0, 0],  # Ground truth aktual
    "timestamp": [datetime.now().isoformat()]  3

})

Simpan ground truth
groundtruthdata.tocsv(

    f"s3://{bucket}/ground-truth/latest.csv",
    index=False
)

from sagemaker.modelmonitor import EndpointInput

Buat endpoint input dengan ground truth
endpointinput = EndpointInput(

    endpointname=predictor.endpointname,

    destination="/opt/ml/processing/input/endpoint",
    inferenceattribute="prediction"
)

Ground truth input
groundtruthinput = f"s3://{bucket}/ground-truth/"

Buat jadwal
modelqualitymonitor.createmonitoringschedule(
    monitorschedulename="model-quality-schedule",
    endpointinput=endpointinput,
    groundtruthinput=groundtruthinput,
    outputs3uri=f"{monitoroutput}/model-quality/reports",

    problemtype="BinaryClassification",
    constraints=modelqualitymonitor.suggestedconstraints(),

    schedulecronexpression=CronExpressionGenerator.daily()

)

from sagemaker.clarify import (
    ModelConfig,
    BiasConfig,
    DataConfig,
    SHAPConfig
)
from sagemaker.modelmonitor import ClarifyModelMonitor

Buat Clarify monitor
clarifymonitor = ClarifyModelMonitor(

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

    volumesizeingb=20,
    maxruntimeinseconds=3600,

    sagemakersession=session
)

Konfigurasi bias
biasconfig = BiasConfig(

    labelvaluesorthreshold=[1],
    facetname="gender",

    facetvaluesorthreshold=[0],  # Monitor bias terhadap gender=0
    groupname="age"

)

Konfigurasi data
dataconfig = DataConfig(
    s3datainputpath=f"s3://{bucket}/training-data/",

    s3outputpath=f"{monitoroutput}/bias/baseline",
    label="target",
    headers=["feature1", "feature2", "gender", "age", "target"],
    datasettype="text/csv"

)

Konfigurasi model
modelconfig = ModelConfig(
    modelname="my-model",

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

    accepttype="text/csv"
)

Buat baseline
clarifymonitor.suggestbaseline(
    dataconfig=dataconfig,
    biasconfig=biasconfig,
    modelconfig=modelconfig,
    wait=True
)

# Buat jadwal bias monitoring
clarifymonitor.createmonitoringschedule(

    monitorschedulename="bias-drift-schedule",

    endpointinput=predictor.endpointname,

    outputs3uri=f"{monitoroutput}/bias/reports",
    constraints=clarifymonitor.suggestedconstraints(),
    schedulecronexpression=CronExpressionGenerator.daily()
)

# Konfigurasi SHAP
shapconfig = SHAPConfig(

    baseline=[
        [0.0]  10  # Nilai baseline untuk setiap fitur
    ],
    numsamples=100,
    aggmethod="meanabs"
)

Buat feature attribution baseline
clarifymonitor.suggestbaseline(
    dataconfig=dataconfig,
    modelconfig=modelconfig,
    explainabilityconfig=shapconfig,
    wait=True
)

2. Jadwalkan Attribution Monitor

# Buat jadwal explainability monitoring
clarifymonitor.createmonitoringschedule(

    monitorschedulename="feature-attribution-schedule",

    endpointinput=predictor.endpointname,

    outputs3uri=f"{monitoroutput}/explainability/reports",
    constraints=clarifymonitor.suggestedconstraints(),
    schedulecronexpression=CronExpressionGenerator.daily(),
    analysistype="explainability"

)

# custommonitor.py
import json
import pandas as pd
import os

def preprocesshandler(inputdata):
    """Preprocess captured data."""
    # Logika preprocessing custom
    return inputdata


def evaluatehandler(baseline, current):
    """Evaluasi data saat ini terhadap baseline."""
    violations = []

    # Cek missing values
    if current.isnull().any().any():
        violations.append({
            "feature": "all",
            "violationtype": "missingvalues",
            "description": "Missing values terdeteksi"
        })

    # Cek distribution shift
    for column in current.columns:
        if column in baseline.columns:
            baselinemean = baseline[column].mean()

            currentmean = current[column].mean()

            if abs(currentmean - baselinemean) > baselinemean * 0.2:

                violations.append({
                    "feature": column,
                    "violationtype": "distributionshift",

                    "description": f"Mean bergeser dari {baselinemean} ke {currentmean}"

                })

    return violations

if name == "main":
    # Load data
    inputpath = "/opt/ml/processing/input"
    outputpath = "/opt/ml/processing/output"


    # Proses dan evaluasi
    # ... implementasi

from sagemaker.modelmonitor import ModelMonitor

Custom monitor dengan BYOC
custommonitor = ModelMonitor(

    role=role,
    imageuri="your-account.dkr.ecr.region.amazonaws.com/custom-monitor:latest",
    instancecount=1,

    instancetype="ml.m5.xlarge",
    volumesizeingb=20,

    maxruntimeinseconds=3600
)

Buat jadwal dengan custom monitor
custommonitor.createmonitoringschedule(

    monitorschedulename="custom-monitor-schedule",

    endpointinput=predictor.endpointname,

    outputs3uri=f"{monitoroutput}/custom/reports",
    schedulecronexpression=CronExpressionGenerator.hourly()
)

Integrasi CloudWatch

1. Lihat Metrics

import boto3
from datetime import datetime, timedelta

cloudwatch = boto3.client("cloudwatch")

Dapatkan metrik
response = cloudwatch.getmetricstatistics(
    Namespace="aws/sagemaker/Endpoints/data-metrics",
    MetricName="featurebaselinedrift",
    Dimensions=[
        {"Name": "Endpoint", "Value": predictor.endpointname},

        {"Name": "MonitoringSchedule", "Value": "data-quality-schedule"}
    ],
    StartTime=datetime.utcnow() - timedelta(hours=24),
    EndTime=datetime.utcnow(),
    Period=3600,
    Statistics=["Average"]
)

for datapoint in response["Datapoints"]:
    print(f"{datapoint['Timestamp']}: {datapoint['Average']}")

# Buat CloudWatch alarm untuk drift
cloudwatch.putmetricalarm(

    AlarmName="ModelDriftAlarm",
    MetricName="featurebaselinedrift",

    Namespace="aws/sagemaker/Endpoints/data-metrics",
    Dimensions=[
        {"Name": "Endpoint", "Value": predictor.endpointname}
    ],
    Statistic="Average",
    Period=3600,
    EvaluationPeriods=2,
    Threshold=0.5,
    ComparisonOperator="GreaterThanThreshold",
    AlarmActions=[
        "arn:aws:sns:region:account:alert-topic"
    ]
)

Analisis Hasil

1. Dapatkan Hasil Monitoring

# List eksekusi monitoring
executions = dataqualitymonitor.listexecutions()


for execution in executions:
    print(f"Eksekusi: {execution.processingjobname}")

    print(f"  Status: {execution.status}")
    print(f"  Mulai: {execution.creationtime}")

Dapatkan eksekusi terbaru
latest = executions[0]
print(f"\nViolations eksekusi terbaru:")

Dapatkan laporan violation
if latest.constraintviolationss3uri:

    violations = latest.constraintviolations()
    for violation in violations.violations:
        print(f"  {violation.featurename}: {violation.constraintchecktype}")

import matplotlib.pyplot as plt

def plotdriftovertime(executions):
    """Plot metrik drift seiring waktu."""
    timestamps = []
    driftscores = []


    for execution in executions:
        if execution.statisticss3uri:

            stats = execution.statistics()
            timestamps.append(execution.creationtime)
            # Ekstrak drift score dari stats
            driftscores.append(stats.get("driftscore", 0))

    plt.figure(figsize=(12, 6))
    plt.plot(timestamps, driftscores, marker='o')

    plt.xlabel("Waktu")
    plt.ylabel("Drift Score")
    plt.title("Model Drift Seiring Waktu")
    plt.xticks(rotation=45)
    plt.tightlayout()
    plt.savefig("driftanalysis.png")


plotdriftovertime(executions)

Best Practices

1. Strategi Monitoring

# Setup monitoring komprehensif
def setupcomprehensivemonitoring(endpointname, baselinedatauri):

    """Setup semua tipe monitor untuk endpoint."""

    monitors = {}

    # 1. Data Quality Monitor
    monitors["dataquality"] = setupdataqualitymonitor(

        endpointname, baselinedatauri
    )

    # 2. Model Quality Monitor (jika ground truth tersedia)
    monitors["modelquality"] = setupmodelqualitymonitor(
        endpointname

    )

    # 3. Bias Monitor
    monitors["bias"] = setupbiasmonitor(endpointname)

    return monitors

2. Konfigurasi Alert

def configuremonitoringalerts(schedulename, snstopicarn):
    """Konfigurasi alert untuk monitoring violations."""

    cloudwatch = boto3.client("cloudwatch")

    # Alert data quality
    cloudwatch.putmetricalarm(

        AlarmName=f"{schedulename}-data-quality-alert",
        MetricName="violationscount",

        Namespace="aws/sagemaker/Endpoints/data-metrics",
        Threshold=1,
        ComparisonOperator="GreaterThanOrEqualToThreshold",
        AlarmActions=[snstopic_arn]
    )

Kesimpulan

SageMaker Model Monitor menyediakan:

Key takeaways:

• Buat baseline dari data training
• Jadwalkan monitoring job secara regular
• Konfigurasi CloudWatch alarms
• Analisis violations dengan segera
• Retrain model ketika drift terdeteksi