SageMaker pipelines

Pipelines automate the lifecycle intermediate

Here's the moment every ML team eventually lives through. A data scientist has a notebook that trains a great model. It works — on their laptop, with their copy of the data, run by their hands in the right order. Then someone asks the obvious question: "Can we retrain it next month on fresh data?" And suddenly nobody's sure. Which cell ran first? Which version of the preprocessing was used? Was that the model we evaluated, or the one before the last tweak? The notebook ran once and produced one model, and now that model is a snowflake — impossible to reproduce, impossible to trust, impossible to hand to anyone else.

Think of the difference between cooking a meal and writing down a recipe. A notebook is the meal: delicious, but gone, and you can only vaguely remember how you made it. A pipeline is the recipe — the exact steps, in order, that anyone (or any schedule) can follow to get the same dish every single time. A SageMaker Pipeline is that recipe expressed in AWS: a fixed, ordered set of steps — preprocess → train → evaluate → register — that AWS runs for you, the same way, on demand.

Now the precise definition, which will land because you feel the pain it solves: a SageMaker Pipeline is a directed acyclic graph (DAG) of steps, where each step's outputs feed the next, and the whole thing is defined as code, versioned, and re-runnable. "Directed acyclic" just means the work flows one way with no loops — data prep can't depend on training that hasn't happened yet. AWS tracks every execution, so you always know exactly which data produced which model, gated by which evaluation. That traceability — not the automation alone — is the real prize. This is the MLOps discipline, expressed in AWS primitives.

The common mistake to name up front: people assume a pipeline is "just my script wrapped in a scheduler." It isn't. The value isn't that it runs unattended — it's that each step is declared, its inputs and outputs are tracked, and the model that comes out the end carries a full lineage back to the data that made it. A cron job running a notebook gives you automation with none of that provenance. A pipeline gives you both, and provenance is what auditors, teammates, and your future self actually need.

A one-off training script is a demo. A SageMaker Pipeline is the automated, repeatable lifecycle: preprocess, train, evaluate, and register — versioned and re-runnable. This is the MLOps discipline, in AWS primitives.

Define the pipeline intermediate

Here's where the mental model usually breaks for newcomers. You sit down to write a pipeline expecting to write the work — the pandas code, the model.fit() call, the loop over epochs. Instead you find yourself writing something that looks oddly empty: a list of step objects with names and instance types, and no actual training logic in sight. It feels like you've written the table of contents but forgotten the book. That confusion is worth resolving, because it reveals what a pipeline definition really is.

A pipeline definition is a plan, not the labour. Picture a general contractor drawing up a build schedule: "first the foundation crew (2 days), then the framers (5 days), then electrical." The contractor isn't pouring concrete while writing the schedule — they're declaring who does what, in what order, on what equipment. When the SageMaker SDK asks you for a ProcessingStep with an SKLearnProcessor on an ml.m5.large, you're doing the same thing: naming a job, saying who runs it and on which cloud machine — not running it.

The two ingredients each step needs are worth separating clearly. The step (a ProcessingStep, a TrainingStep) is the what — a named box in the graph. The processor or estimator handed to it is the who and where — which container image does the work and what size instance it runs on. Keeping those two ideas distinct is the whole reason the API has two kinds of object: the step is the node in the DAG; the processor/estimator is the worker assigned to that node.

And the ordering rule is simpler than people fear: the order you list steps in is the order they run. When you write steps=[Preprocess, Train], you've declared that training depends on preprocessing finishing first. (In richer pipelines the dependency comes from one step consuming another's output — passing a training step the artifact a processing step produced — but the intuition is the same: outputs flow forward.) Finally, note the verb at the end: upsert — a blend of "update" and "insert." It ships the definition to AWS and saves the recipe; it does not cook. Nothing trains until a separate call starts an execution. The frequent beginner mistake is expecting upsert to kick off a run and then wondering why no model appeared — it only registered the plan.

pipeline.pyfrom sagemaker.workflow.pipeline import Pipeline
from sagemaker.workflow.steps import ProcessingStep, TrainingStep
from sagemaker.sklearn.processing import SKLearnProcessor
from sagemaker.estimator import Estimator

ROLE = "arn:aws:iam::123456789012:role/sm-pipeline-role"

# Build a processor and an estimator (real args elided), then wire them
# into a repeatable DAG. This is the whole point of a Pipeline.
processor = SKLearnProcessor(framework_version="1.2-1", role=ROLE, instance_count=1,
                             instance_type="ml.m5.large")
estimator = Estimator(image_uri="<training-image>", role=ROLE, instance_count=1,
                      instance_type="ml.m5.large")

pipeline = Pipeline(
    name="churn-model-pipeline",
    steps=[
        ProcessingStep(name="Preprocess", processor=processor),
        TrainingStep(name="Train", estimator=estimator),
    ],
)
pipeline.upsert(role_arn=ROLE)

Model Registry advanced

Ask any team that has shipped ML to production about their worst afternoon, and a version of this story comes up. Model quality drops in production. Someone needs to roll back to "the good one." But which artifact was the good one? There are seven .tar.gz files in an S3 bucket named things like model-final, model-final-v2, and model-really-final. Nobody recorded which one scored what, or which one is actually live. The rollback becomes archaeology. This is the exact chaos the Model Registry exists to prevent.

Think of it as the version control for models — a Git-like ledger, but for trained artifacts instead of source. Every time a model is registered it gets a numbered version in a named group (e.g. version 3 in the churn group), and alongside the artifact the registry stores its metrics and, crucially, a status field called ModelApprovalStatus. That one field is the star of this section, so hold onto it.

Here's the precise idea and why it matters: ModelApprovalStatus is a gate, and the deploy step is built to read it. A version can sit at PendingManualApproval, and be flipped to Approved or Rejected. Your deployment is configured to ship only Approved versions — so promotion is never automatic on faith. A human, or an automated check, must set the status before the model can reach production. This decouples "a model exists and passed evaluation" from "a model is cleared to serve traffic" — two genuinely different decisions that beginners often collapse into one.

The subtle point people miss: the registry doesn't make the decision — it records and enforces it. Training and evaluation can happen fully automatically inside the pipeline; the approval flip is a deliberate, separate, auditable act. That separation is the backbone of an eval-gated rollout: the pipeline can register a candidate the moment it beats a baseline, but the candidate waits at the gate until someone opens it. It's the same discipline you met in O3 — and it's why "who approved this model, and when?" always has a clean answer.

Training outputs a model artifact; the Model Registry versions it, tracks metrics, and gates promotion with an approval status. Your deploy step only ships Approved versions — an eval-gated rollout, exactly like O3.

approve.pyimport boto3
sm = boto3.client("sagemaker", region_name="us-east-1")

# promote a specific model version to Approved so deployment can pick it up
sm.update_model_package(
    ModelPackageArn="arn:aws:sagemaker:us-east-1:123456789012:model-package/churn/3",
    ModelApprovalStatus="Approved",
)

Run and inspect expert

You've written the recipe and saved it. Now comes a distinction that trips people up in every distributed system: defining the work and doing the work are two different acts. Your pipeline exists in AWS, versioned and ready — but it is idle. It has never trained anything. Nothing happens until you explicitly press Start. And once you do, a second surprise waits: the call that starts the run returns almost instantly, long before any training finishes.

That's because a pipeline run is asynchronous. Think of ordering food at a busy kitchen: you place the order and immediately get a ticket number — you don't stand frozen at the counter until the meal is cooked. start_pipeline_execution is placing the order; the ticket it hands back is the execution ARN, a unique ID for this specific run. AWS then works through your steps in the background while your script has already moved on.

This is why "run" and "inspect" are naturally two separate operations, and why one call cannot do both. To learn how the run is going, you take that ticket number and ask a different question: list_pipeline_execution_steps, passing the execution ARN. It reports each step and its current status — Executing, Succeeded, Failed. Because the run is live, that status is a snapshot in time: ask again a minute later and a step that was Executing may now read Succeeded. The mental model is a pizza-tracker — the order is placed once, but you refresh the status page repeatedly to watch it move.

The common mistake here is treating a single status check as the answer. One glance tells you where things stand right now, not whether the run will ultimately succeed. Real monitoring means polling — re-querying on an interval until every step reaches a terminal state (that's exactly what the practice ladder builds toward). And one gotcha to file away: for a brief moment right after you start a run, the step list can come back empty before AWS has registered the steps — so robust polling code checks for a non-empty list before deciding the run is done, or it will exit early on the first tick.

run_pipeline.pyimport boto3
sm = boto3.client("sagemaker", region_name="us-east-1")

ex = sm.start_pipeline_execution(PipelineName="churn-model-pipeline")
print("execution:", ex["PipelineExecutionArn"])

steps = sm.list_pipeline_execution_steps(PipelineExecutionArn=ex["PipelineExecutionArn"])
for s in steps["PipelineExecutionSteps"]:
    print(s["StepName"], s["StepStatus"])

preprocess -> train -> evaluate -> register

from sagemaker.workflow.pipeline import Pipeline
from sagemaker.workflow.steps import ProcessingStep, TrainingStep
from sagemaker.sklearn.processing import SKLearnProcessor
from sagemaker.estimator import Estimator

ROLE = "arn:aws:iam::123456789012:role/sm-pipeline-role"

processor = SKLearnProcessor(framework_version="1.2-1", role=ROLE, instance_count=1,
                             instance_type="ml.m5.large")
estimator = Estimator(image_uri="<training-image>", role=ROLE, instance_count=1,
                      instance_type="ml.m5.large")

pipeline = Pipeline(
    name="churn-model-pipeline",
    steps=[
        ProcessingStep(name="Preprocess", processor=processor),
        TrainingStep(name="Train", estimator=estimator),
    ],
)
pipeline.upsert(role_arn=ROLE)

import boto3
sm = boto3.client("sagemaker", region_name="us-east-1")

# promote a specific model version to Approved so deployment can pick it up
sm.update_model_package(
    ModelPackageArn="arn:aws:sagemaker:us-east-1:123456789012:model-package/churn/3",
    ModelApprovalStatus="Approved",
)

import boto3
sm = boto3.client("sagemaker", region_name="us-east-1")

ex = sm.start_pipeline_execution(PipelineName="churn-model-pipeline")
print("execution:", ex["PipelineExecutionArn"])

steps = sm.list_pipeline_execution_steps(PipelineExecutionArn=ex["PipelineExecutionArn"])
for s in steps["PipelineExecutionSteps"]:
    print(s["StepName"], s["StepStatus"])

import boto3, time
sm = boto3.client("sagemaker", region_name="us-east-1")

ex = sm.start_pipeline_execution(PipelineName="churn-model-pipeline")
arn = ex["PipelineExecutionArn"]
print("execution:", arn)

TERMINAL = {"Succeeded", "Failed", "Stopped"}
while True:
    steps = sm.list_pipeline_execution_steps(PipelineExecutionArn=arn)["PipelineExecutionSteps"]
    statuses = {s["StepName"]: s["StepStatus"] for s in steps}
    print(statuses)
    if steps and all(v in TERMINAL for v in statuses.values()):
        break
    time.sleep(30)

print("done:", "FAILED" if "Failed" in statuses.values() else "ALL SUCCEEDED")

from sagemaker.workflow.pipeline import Pipeline
from sagemaker.workflow.steps import ProcessingStep, TrainingStep
from sagemaker.workflow.condition_step import ConditionStep
from sagemaker.workflow.conditions import ConditionGreaterThanOrEqualTo
from sagemaker.workflow.step_collections import RegisterModel

ROLE = "arn:aws:iam::123456789012:role/sm-pipeline-role"

# ... processor, estimator, and an 'evaluate' ProcessingStep that emits accuracy ...
register = RegisterModel(
    name="RegisterChurn",
    estimator=estimator,
    model_package_group_name="churn",
    approval_status="PendingManualApproval",   # not auto-shipped
)
gate = ConditionStep(
    name="GateOnAccuracy",
    conditions=[ConditionGreaterThanOrEqualTo(left=eval_accuracy, right=0.80)],
    if_steps=[register],          # only register if it beats the 0.80 baseline
    else_steps=[],
)
pipeline = Pipeline(name="churn-model-pipeline",
                    steps=[preprocess, train, evaluate, gate])
pipeline.upsert(role_arn=ROLE)

import boto3
sm = boto3.client("sagemaker", region_name="us-east-1")
sm.update_model_package(
    ModelPackageArn="arn:aws:sagemaker:us-east-1:123456789012:model-package/churn/3",
    ModelApprovalStatus="Approved",
)

Knowledge check check yourself