The smallest MLOps setup worth having on SageMaker AI is a pipeline that preprocesses, trains, evaluates, and registers a model only when it clears a quality bar, plus an event that deploys whatever gets approved. This tutorial builds exactly that, end to end, with a runnable pipeline definition. Everything uses the SageMaker Python SDK (2.200 or newer) and XGBoost so it runs on cheap CPU instances.
The loop
S3 raw data --> Preprocess --> Train --> Evaluate --> (accuracy >= threshold?) --> Register
|
EventBridge (approval status changed) <----+--> Deploy
Three scripts and one pipeline definition. Put the scripts in a src/ directory next to pipeline.py.
Step 1: preprocessing script
src/preprocess.py reads a CSV, splits it, and writes train / validation / test sets in the headerless format XGBoost expects (label first):
import pandas as pd
from sklearn.model_selection import train_test_split
df = pd.read_csv("/opt/ml/processing/input/data.csv")
y = df.pop("label")
train_x, rest_x, train_y, rest_y = train_test_split(df, y, test_size=0.3, random_state=42)
val_x, test_x, val_y, test_y = train_test_split(rest_x, rest_y, test_size=0.5, random_state=42)
for name, x, y_ in [("train", train_x, train_y), ("validation", val_x, val_y), ("test", test_x, test_y)]:
out = pd.concat([y_.reset_index(drop=True), x.reset_index(drop=True)], axis=1)
out.to_csv(f"/opt/ml/processing/{name}/{name}.csv", header=False, index=False)
Step 2: evaluation script
src/evaluate.py loads the trained model, scores the test set, and writes a JSON report the pipeline can read:
import json, tarfile
import pandas as pd
import xgboost as xgb
from sklearn.metrics import accuracy_score, roc_auc_score
with tarfile.open("/opt/ml/processing/model/model.tar.gz") as t:
t.extractall("/tmp/model")
booster = xgb.Booster()
booster.load_model("/tmp/model/xgboost-model")
test = pd.read_csv("/opt/ml/processing/test/test.csv", header=None)
y, x = test.iloc[:, 0], test.iloc[:, 1:]
probs = booster.predict(xgb.DMatrix(x))
preds = (probs > 0.5).astype(int)
report = {
"binary_classification_metrics": {
"accuracy": {"value": float(accuracy_score(y, preds))},
"auc": {"value": float(roc_auc_score(y, probs))},
}
}
with open("/opt/ml/processing/evaluation/evaluation.json", "w") as f:
json.dump(report, f)
The nested binary_classification_metrics shape is the Model Registry's model-quality format, so the same file doubles as the registered model's metrics card.
Step 3: the pipeline definition
pipeline.py:
import sagemaker
from sagemaker.workflow.pipeline import Pipeline
from sagemaker.workflow.pipeline_context import PipelineSession
from sagemaker.workflow.parameters import ParameterString, ParameterFloat
from sagemaker.workflow.steps import ProcessingStep, TrainingStep
from sagemaker.workflow.model_step import ModelStep
from sagemaker.workflow.condition_step import ConditionStep
from sagemaker.workflow.conditions import ConditionGreaterThanOrEqualTo
from sagemaker.workflow.functions import JsonGet, Join
from sagemaker.workflow.properties import PropertyFile
from sagemaker.sklearn.processing import SKLearnProcessor
from sagemaker.processing import ProcessingInput, ProcessingOutput
from sagemaker.estimator import Estimator
from sagemaker.inputs import TrainingInput
from sagemaker.model import Model
from sagemaker.model_metrics import ModelMetrics, MetricsSource
session = PipelineSession()
role = sagemaker.get_execution_role()
region = session.boto_region_name
bucket = session.default_bucket()
input_data = ParameterString("InputData", default_value=f"s3://{bucket}/raw/data.csv")
accuracy_threshold = ParameterFloat("AccuracyThreshold", default_value=0.85)
approval_status = ParameterString("ModelApprovalStatus", default_value="PendingManualApproval")
# --- Preprocess ---
sklearn = SKLearnProcessor(
framework_version="1.2-1", role=role, instance_type="ml.m5.large",
instance_count=1, sagemaker_session=session,
)
preprocess = ProcessingStep(
name="Preprocess",
step_args=sklearn.run(
code="src/preprocess.py",
inputs=[ProcessingInput(source=input_data, destination="/opt/ml/processing/input")],
outputs=[
ProcessingOutput(output_name="train", source="/opt/ml/processing/train"),
ProcessingOutput(output_name="validation", source="/opt/ml/processing/validation"),
ProcessingOutput(output_name="test", source="/opt/ml/processing/test"),
],
),
)
# --- Train ---
xgb_image = sagemaker.image_uris.retrieve("xgboost", region, version="1.7-1")
xgb = Estimator(
image_uri=xgb_image, role=role, instance_type="ml.m5.xlarge", instance_count=1,
output_path=f"s3://{bucket}/models", sagemaker_session=session,
)
xgb.set_hyperparameters(objective="binary:logistic", num_round=200, max_depth=5, eta=0.2)
train = TrainingStep(
name="Train",
step_args=xgb.fit({
"train": TrainingInput(
preprocess.properties.ProcessingOutputConfig.Outputs["train"].S3Output.S3Uri,
content_type="text/csv"),
"validation": TrainingInput(
preprocess.properties.ProcessingOutputConfig.Outputs["validation"].S3Output.S3Uri,
content_type="text/csv"),
}),
)
# --- Evaluate ---
evaluation_report = PropertyFile(name="EvaluationReport", output_name="evaluation", path="evaluation.json")
evaluate = ProcessingStep(
name="Evaluate",
step_args=sklearn.run(
code="src/evaluate.py",
inputs=[
ProcessingInput(source=train.properties.ModelArtifacts.S3ModelArtifacts,
destination="/opt/ml/processing/model"),
ProcessingInput(source=preprocess.properties.ProcessingOutputConfig.Outputs["test"].S3Output.S3Uri,
destination="/opt/ml/processing/test"),
],
outputs=[ProcessingOutput(output_name="evaluation", source="/opt/ml/processing/evaluation")],
),
property_files=[evaluation_report],
)
# --- Register (only if the condition passes) ---
model = Model(image_uri=xgb_image, model_data=train.properties.ModelArtifacts.S3ModelArtifacts,
role=role, sagemaker_session=session)
metrics = ModelMetrics(model_statistics=MetricsSource(
s3_uri=Join(on="/", values=[
evaluate.properties.ProcessingOutputConfig.Outputs["evaluation"].S3Output.S3Uri,
"evaluation.json",
]),
content_type="application/json",
))
register = ModelStep(
name="Register",
step_args=model.register(
content_types=["text/csv"], response_types=["text/csv"],
inference_instances=["ml.m5.large"], transform_instances=["ml.m5.large"],
model_package_group_name="churn-xgboost",
approval_status=approval_status,
model_metrics=metrics,
),
)
condition = ConditionStep(
name="AccuracyGate",
conditions=[ConditionGreaterThanOrEqualTo(
left=JsonGet(step_name=evaluate.name, property_file=evaluation_report,
json_path="binary_classification_metrics.accuracy.value"),
right=accuracy_threshold,
)],
if_steps=[register],
else_steps=[],
)
pipeline = Pipeline(
name="churn-xgboost-pipeline",
parameters=[input_data, accuracy_threshold, approval_status],
steps=[preprocess, train, evaluate, condition],
sagemaker_session=session,
)
if __name__ == "__main__":
pipeline.upsert(role_arn=role)
execution = pipeline.start()
print(execution.arn)
Pipeline properties are resolved at run time, so you cannot string-concatenate them; that is why the metrics S3 path is built with Join rather than an f-string.
Run python pipeline.py. The first execution takes roughly ten minutes on the instance sizes above. Inspect it in Studio under Pipelines, or with execution.list_steps().
Step 4: conditional registration in practice
The ConditionStep is the whole point of the loop. A run that trains a worse model than the threshold simply ends without registering anything, which keeps the Model Registry a list of candidates worth looking at. Tune AccuracyThreshold per execution without editing code:
pipeline.start(parameters={"AccuracyThreshold": 0.9})
Registered packages land in the churn-xgboost model package group with PendingManualApproval. A reviewer (or an automated check) flips the status:
import boto3
sm = boto3.client("sagemaker")
sm.update_model_package(
ModelPackageArn=package_arn,
ModelApprovalStatus="Approved",
ApprovalDescription="Beat prod AUC by 0.02 on the August test set",
)
Step 5: EventBridge-triggered deploy
Approval emits a SageMaker Model Package State Change event. An EventBridge rule routes it to a Lambda function that creates or updates the endpoint:
{
"source": ["aws.sagemaker"],
"detail-type": ["SageMaker Model Package State Change"],
"detail": {
"ModelPackageGroupName": ["churn-xgboost"],
"ModelApprovalStatus": ["Approved"]
}
}
The Lambda (Python, boto3):
import os, time, boto3
sm = boto3.client("sagemaker")
ROLE = os.environ["SAGEMAKER_ROLE_ARN"]
ENDPOINT = "churn-xgboost-prod"
def handler(event, _context):
package_arn = event["detail"]["ModelPackageArn"]
stamp = str(int(time.time()))
model_name = f"churn-{stamp}"
config_name = f"churn-config-{stamp}"
sm.create_model(
ModelName=model_name, ExecutionRoleArn=ROLE,
PrimaryContainer={"ModelPackageName": package_arn},
)
sm.create_endpoint_config(
EndpointConfigName=config_name,
ProductionVariants=[{
"VariantName": "AllTraffic", "ModelName": model_name,
"InitialInstanceCount": 1, "InstanceType": "ml.m5.large",
}],
)
existing = sm.list_endpoints(NameContains=ENDPOINT)["Endpoints"]
if any(e["EndpointName"] == ENDPOINT for e in existing):
sm.update_endpoint(EndpointName=ENDPOINT, EndpointConfigName=config_name)
else:
sm.create_endpoint(EndpointName=ENDPOINT, EndpointConfigName=config_name)
return {"deployed": package_arn}
update_endpoint performs a blue / green swap with no downtime. Put the Lambda, the rule, and the IAM roles in CDK or Terraform so the deploy path is reviewable and reproducible, and add a manual approval stage in front of it for production if your change process needs one.
What to add next
- Data capture and Model Monitor on the endpoint, so the loop closes with drift alerts.
- A retraining schedule (EventBridge Scheduler calling
pipeline.start()) or a trigger on new data landing in S3. - Pipeline tests:
pipeline.definition()returns JSON you can lint and diff in CI beforeupsert. - Shadow or canary variants on the endpoint before sending 100 percent of traffic to a new package.
This loop is the foundation of every MLOps engagement we run; see our MLOps consulting page for what the production-grade version adds. Questions about your own pipeline? Contact us.