Computer Pathshalaकंप्यूटर पाठशाला
Start free
videoLesson 02· 28 minFree previewContent ready

Your first SageMaker pipeline — train → register → deploy

Video coming soon. The lesson script and lab are ready; recording is in production.voiceover script available ↓

Your first SageMaker pipeline — train → register → deploy

We've spent Lesson 01 establishing why most ML projects die in the engineering work around the model. This lesson is the antidote in miniature. By the end you'll have seen the exact three-stage backbone every production ML system on SageMaker uses: train a model, register it as a versioned, governable artifact, and deploy it to a live endpoint with shadow traffic. Every later week of the course is a deeper dive into one of these stages — but if you only watched one lesson, this is the one that would make the rest make sense.

We'll build it end-to-end, in code, with no clicking through the AWS console. The pattern is the one we ship for client projects at Computer Pathshala. By the time you finish the lab at the end of the week, you'll have run a complete version of it in your own AWS sandbox.

The three-stage backbone

Every box on that diagram corresponds to a SageMaker concept we'll wire up in code:

Why we don't just pickle.dump(model)

If you've trained an XGBoost model in a notebook, you might be wondering why we need any of this machinery at all. Can't we just save the model with pickle.dump(model, open("model.pkl", "wb")) and serve it from a Flask app? Yes, you can. Many teams do, in fact, do exactly that. Three months later, they have these problems:

The SageMaker backbone we're building gives you all four of those properties for free. That's why it's worth the upfront investment.

Setting up — what you need

This lesson assumes:

If you don't have those yet, complete the lab at the end of this week first — it walks you through spinning up the SageMaker domain — and come back to this lesson.

Step 1 — define the pipeline parameters

We start by declaring the parameters of the pipeline. These are values you'll pass in when you run the pipeline — not hard-coded constants in your notebook. Parameters are what lets the same pipeline definition serve many runs (different data versions, different hyperparameters, different output paths) without editing the code.

# pipeline.py
import sagemaker
from sagemaker.workflow.parameters import (
    ParameterString,
    ParameterFloat,
    ParameterInteger,
)

# What can vary between pipeline runs:
input_data = ParameterString(
    name="InputData",
    default_value="s3://cp-mlops-sandbox/raw/abalone-train.csv",
)
model_approval_status = ParameterString(
    name="ModelApprovalStatus",
    default_value="PendingManualApproval",
)
accuracy_threshold = ParameterFloat(
    name="AccuracyThreshold",
    default_value=0.75,
)
training_instance_type = ParameterString(
    name="TrainingInstanceType",
    default_value="ml.m5.large",
)
training_instance_count = ParameterInteger(
    name="TrainingInstanceCount",
    default_value=1,
)

Note model_approval_status defaults to "PendingManualApproval". Even when the model passes the accuracy threshold, it doesn't auto-deploy to customers — it sits in the registry waiting for an approver. We'll change that in week 8 once we have a CI pipeline that earns auto-approval. For now, manual approval is the right default.

Step 2 — the training step

The training step runs a SageMaker Estimator. We'll use the built-in XGBoost container so we don't have to bring our own image yet (week 4 covers BYOC). For a regression task, our Estimator looks like this:

from sagemaker.estimator import Estimator
from sagemaker.workflow.steps import TrainingStep
from sagemaker.inputs import TrainingInput
from sagemaker.image_uris import retrieve

role = sagemaker.get_execution_role()
sagemaker_session = sagemaker.Session()
default_bucket = sagemaker_session.default_bucket()

xgboost_image = retrieve(
    framework="xgboost",
    region=sagemaker_session.boto_region_name,
    version="1.7-1",
)

xgb_train = Estimator(
    image_uri=xgboost_image,
    instance_type=training_instance_type,
    instance_count=training_instance_count,
    output_path=f"s3://{default_bucket}/mlops-pilot/xgb-output",
    role=role,
)
xgb_train.set_hyperparameters(
    objective="reg:squarederror",
    num_round=50,
    max_depth=5,
    eta=0.2,
    gamma=4,
    min_child_weight=6,
    subsample=0.7,
)

step_train = TrainingStep(
    name="TrainAbaloneXGB",
    estimator=xgb_train,
    inputs={
        "train": TrainingInput(s3_data=input_data, content_type="text/csv"),
    },
)

Three things to notice:

  1. We retrieve the XGBoost image URI for our specific region instead of hard-coding it. AWS image registries are regional; hard-coded URIs are the most common cause of "works in my notebook, fails in CI" bugs.
  2. The estimator's output path lives in the SageMaker default bucket (sagemaker-<region>-<accountid>). Don't put model artifacts in a custom bucket without thinking — the default bucket has the IAM policies already wired for SageMaker to write into it.
  3. The TrainingStep accepts the parameter object (input_data) directly. That's what makes the pipeline parameterisable.

Step 3 — the evaluation step

After training, we run a small Python script that loads the trained model, runs it on a held-out test set, and computes accuracy (or RMSE for our regression problem). The script writes the result to a JSON file that the next step can read.

# evaluate.py — runs inside the SageMaker Processing container
import json
import pathlib
import pickle
import tarfile

import pandas as pd
import xgboost as xgb
from sklearn.metrics import mean_squared_error


def main():
    model_path = "/opt/ml/processing/model/model.tar.gz"
    with tarfile.open(model_path) as tf:
        tf.extractall(path=".")

    booster = pickle.load(open("xgboost-model", "rb"))

    test_df = pd.read_csv("/opt/ml/processing/test/test.csv", header=None)
    y_test = test_df.iloc[:, 0].to_numpy()
    X_test = xgb.DMatrix(test_df.iloc[:, 1:].to_numpy())
    predictions = booster.predict(X_test)

    rmse = float(mean_squared_error(y_test, predictions) ** 0.5)
    report = {"regression_metrics": {"rmse": {"value": rmse}}}

    output_dir = "/opt/ml/processing/evaluation"
    pathlib.Path(output_dir).mkdir(parents=True, exist_ok=True)
    with open(f"{output_dir}/evaluation.json", "w") as f:
        json.dump(report, f)


if __name__ == "__main__":
    main()

The script runs inside a SageMaker ProcessingStep:

from sagemaker.processing import ScriptProcessor, ProcessingInput, ProcessingOutput
from sagemaker.workflow.steps import ProcessingStep
from sagemaker.workflow.properties import PropertyFile

eval_processor = ScriptProcessor(
    image_uri=xgboost_image,
    command=["python3"],
    instance_type="ml.m5.large",
    instance_count=1,
    role=role,
)

evaluation_report = PropertyFile(
    name="EvaluationReport",
    output_name="evaluation",
    path="evaluation.json",
)

step_eval = ProcessingStep(
    name="EvaluateAbaloneXGB",
    processor=eval_processor,
    inputs=[
        ProcessingInput(
            source=step_train.properties.ModelArtifacts.S3ModelArtifacts,
            destination="/opt/ml/processing/model",
        ),
        ProcessingInput(
            source=input_data,
            destination="/opt/ml/processing/test",
        ),
    ],
    outputs=[
        ProcessingOutput(output_name="evaluation", source="/opt/ml/processing/evaluation"),
    ],
    code="evaluate.py",
    property_files=[evaluation_report],
)

The PropertyFile is the bridge between the evaluation step's output and the next step's condition. SageMaker parses the JSON file at the given path and exposes its values to the rest of the pipeline.

Step 4 — conditional registration

Now the safety gate. We register the model only if RMSE is below a threshold. If the model is worse than that, the pipeline fails and the model never enters the registry.

from sagemaker.workflow.conditions import ConditionLessThanOrEqualTo
from sagemaker.workflow.condition_step import ConditionStep
from sagemaker.workflow.functions import JsonGet
from sagemaker.workflow.model_step import ModelStep
from sagemaker.model import Model

model = Model(
    image_uri=xgboost_image,
    model_data=step_train.properties.ModelArtifacts.S3ModelArtifacts,
    sagemaker_session=sagemaker_session,
    role=role,
)

register_step = ModelStep(
    name="RegisterAbaloneXGB",
    step_args=model.register(
        content_types=["text/csv"],
        response_types=["text/csv"],
        inference_instances=["ml.t2.medium", "ml.m5.large"],
        transform_instances=["ml.m5.large"],
        model_package_group_name="AbaloneXGBPackageGroup",
        approval_status=model_approval_status,
    ),
)

cond_metrics = ConditionLessThanOrEqualTo(
    left=JsonGet(
        step_name=step_eval.name,
        property_file=evaluation_report,
        json_path="regression_metrics.rmse.value",
    ),
    # We want LOW RMSE; "accuracy_threshold" is a misnomer for regression but
    # the parameter is reused for both task types in this course's sandbox.
    right=accuracy_threshold,
)

step_cond = ConditionStep(
    name="CheckRMSE",
    conditions=[cond_metrics],
    if_steps=[register_step],
    else_steps=[],   # fall through — pipeline fails to register
)

Step 5 — assemble and run

Finally, we wire everything together:

from sagemaker.workflow.pipeline import Pipeline

pipeline = Pipeline(
    name="AbaloneMLOpsPilot",
    parameters=[
        input_data,
        model_approval_status,
        accuracy_threshold,
        training_instance_type,
        training_instance_count,
    ],
    steps=[step_train, step_eval, step_cond],
    sagemaker_session=sagemaker_session,
)

pipeline.upsert(role_arn=role)
execution = pipeline.start()
execution.wait()

The first pipeline.upsert() registers (or updates) the pipeline definition in SageMaker. The start() kicks off an execution and wait() blocks until it completes (success or failure). In CI we don't wait — we hand off the execution ARN to a downstream step. For now, blocking is fine.

Step 6 — deploy from the registry

After the pipeline succeeds, the model package sits in the registry with status PendingManualApproval. To deploy it, an approver (a human, or in week 8 a CI bot) flips it to Approved:

import boto3
sm = boto3.client("sagemaker")

# Find the latest model package
group = "AbaloneXGBPackageGroup"
packages = sm.list_model_packages(
    ModelPackageGroupName=group,
    SortBy="CreationTime",
    SortOrder="Descending",
    MaxResults=1,
)
latest = packages["ModelPackageSummaryList"][0]["ModelPackageArn"]

sm.update_model_package(
    ModelPackageArn=latest,
    ModelApprovalStatus="Approved",
    ApprovalDescription="Manual approval — Week 1 pilot",
)

Then we deploy:

from sagemaker import ModelPackage

approved_model = ModelPackage(
    role=role,
    model_package_arn=latest,
    sagemaker_session=sagemaker_session,
)
approved_model.deploy(
    initial_instance_count=1,
    instance_type="ml.t2.medium",
    endpoint_name="abalone-mlops-pilot",
)

Within ~5 minutes you have a live HTTPS endpoint serving the model. Send it a CSV row, get back a prediction:

predictor = sagemaker.predictor.Predictor(endpoint_name="abalone-mlops-pilot")
predictor.serializer = sagemaker.serializers.CSVSerializer()
predictor.deserializer = sagemaker.deserializers.CSVDeserializer()
print(predictor.predict("M,0.455,0.365,0.095,0.514,0.2245,0.101,0.15"))

Where we'll add shadow traffic

The pattern as written deploys the new model to 100% of traffic. That's fine for a pilot, but it's not how you'd promote a model in production — you'd want to compare it against the current production model on real traffic before flipping the switch.

In week 6, we'll modify the deployment step to create a production variant with InitialVariantWeight=0 and a shadow variant that gets traffic mirrored to it. The current production model keeps serving customers; the new model gets the same requests in parallel; we compare their responses. Only after we're satisfied do we shift traffic weight to the new variant.

For now, just remember: deploying to an endpoint is not the same as promoting to production. The pattern matters, even if we're elliding the detail until week 6.

Key takeaways

Lab

The end-of-week lab walks you through running this exact pipeline in the course sandbox. By the time you finish it, you'll have an AbaloneXGB model package in your registry, an abalone-mlops-pilot endpoint deployed, and the muscle memory to recreate the pattern on the project of your choice.

What's next

In Lesson 03 we'll zoom out from the pipeline code itself and map the "boring 70%" — the data, eval, monitoring and ownership work — onto the SageMaker services that solve each piece. You'll come away with the mental model that lets you read any AWS reference architecture for ML and know which boxes you've already learned to draw and which boxes are upcoming weeks.