SageMaker Pipelines: Automating the ML Lifecycle

This is the fourth post in our Amazon SageMaker series, building on notebooks and training jobs, built-in algorithms, and deployment endpoints.

Every step covered so far in this series — preprocessing, training, evaluation, deployment — has been a manually-run notebook cell, executed by a person, in an order they have to remember correctly, using the exact right parameters each time. This is the same fragile, undocumented, "hope the person doing it remembers the steps" pattern this blog has argued against repeatedly, for Terraform applied by hand, for a deploy process run manually before we automated it with GitHub Actions in our own real case study. SageMaker Pipelines is the fix for the ML lifecycle specifically. Because a real production ML pipeline touches more distinct concerns than almost anything else covered in this series — data processing, training, conditional quality gates, versioning, and scheduling — this closing technical post goes deeper than the rest.

Defining a pipeline as a sequence of steps

from sagemaker.workflow.pipeline import Pipeline
from sagemaker.workflow.steps import ProcessingStep, TrainingStep
from sagemaker.workflow.step_collections import RegisterModel
from sagemaker.processing import ScriptProcessor
 
processor = ScriptProcessor(
    image_uri="...",
    role=role,
    instance_count=1,
    instance_type="ml.m5.xlarge",
)
 
preprocessing_step = ProcessingStep(
    name="PreprocessData",
    processor=processor,
    inputs=[...],
    outputs=[...],
    code="preprocessing.py",
)
 
training_step = TrainingStep(
    name="TrainModel",
    estimator=estimator,
    inputs={
        "train": preprocessing_step.properties.ProcessingOutputConfig.Outputs["train"].S3Output.S3Uri,
    },
)
 
register_step = RegisterModel(
    name="RegisterModel",
    estimator=estimator,
    model_data=training_step.properties.ModelArtifacts.S3ModelArtifacts,
    content_types=["text/csv"],
    response_types=["text/csv"],
    model_package_group_name="churn-prediction-models",
)
 
pipeline = Pipeline(
    name="churn-prediction-pipeline",
    steps=[preprocessing_step, training_step, register_step],
)
 
pipeline.upsert(role_arn=role)
pipeline.start()

This is genuinely the same conceptual shape as a CI/CD workflow — a defined sequence of steps, each depending on the previous step's output, executed automatically rather than by a person manually clicking through notebook cells in order. preprocessing_step.properties... referencing the previous step's actual output is the pipeline equivalent of a job's needs: dependency in GitHub Actions, covered directly earlier in this blog: an explicit, code-defined dependency, not an implicit assumption about execution order that only exists in someone's memory.

Conditional steps: only registering a model that's actually good enough

from sagemaker.workflow.condition_step import ConditionStep
from sagemaker.workflow.conditions import ConditionGreaterThanOrEqualTo
from sagemaker.workflow.functions import JsonGet
 
evaluation_step = ProcessingStep(
    name="EvaluateModel",
    processor=processor,
    inputs=[...],
    outputs=[...],
    code="evaluate.py",
)
 
condition = ConditionGreaterThanOrEqualTo(
    left=JsonGet(step_name=evaluation_step.name, property_file="evaluation.json", json_path="metrics.auc"),
    right=0.75,
)
 
condition_step = ConditionStep(
    name="CheckModelQuality",
    conditions=[condition],
    if_steps=[register_step],
    else_steps=[],
)

This is the ML pipeline equivalent of a quality gate in any other CI/CD process — the same instinct behind requiring tests to pass before a deploy proceeds, or requiring a Molecule test to pass before an Ansible role change ships, covered earlier in this blog. A newly-trained model only gets registered (and becomes eligible for deployment) if its evaluated AUC meets a defined bar — a model that trained "successfully" but performs worse than the bar never reaches production, automatically, with no person needing to remember to manually check a metric before deciding whether to deploy.

Comparing against the currently deployed model, not just a fixed threshold

A fixed threshold (AUC >= 0.75) is a reasonable starting gate, but a more rigorous check compares a newly-trained candidate directly against whatever model is currently deployed — a model can clear a fixed threshold while still being a regression relative to what's already in production:

condition = ConditionGreaterThan(
    left=JsonGet(step_name=evaluation_step.name, property_file="evaluation.json", json_path="metrics.auc"),
    right=JsonGet(step_name=get_current_model_metrics_step.name, property_file="current_metrics.json", json_path="metrics.auc"),
)

This requires an extra step that looks up the currently-deployed model's own recorded evaluation metrics (from the Model Registry, covered below) before comparing — genuinely more work to set up than a fixed threshold, and a meaningfully stronger guarantee: a candidate model only replaces production if it's actually better than what's already there, not merely "good enough" in isolation.

Parameterizing a pipeline for reuse across environments

The same "don't hardcode values a caller should be able to vary" principle covered for Terraform modules and Ansible roles applies directly to pipelines:

from sagemaker.workflow.parameters import ParameterString, ParameterFloat
 
input_data_uri = ParameterString(name="InputDataUri", default_value=f"s3://{bucket}/training-data/")
auc_threshold = ParameterFloat(name="AucThreshold", default_value=0.75)
 
pipeline = Pipeline(
    name="churn-prediction-pipeline",
    parameters=[input_data_uri, auc_threshold],
    steps=[preprocessing_step, training_step, evaluation_step, condition_step],
)
 
pipeline.start(parameters={"InputDataUri": f"s3://{bucket}/training-data-v2/", "AucThreshold": 0.8})

Parameterizing the pipeline's data source and quality threshold means the exact same pipeline definition can run against different datasets or under a stricter quality bar for a specific environment (a staging pipeline run with a lower threshold for faster iteration, a production pipeline run with a stricter one) without duplicating the pipeline definition itself — the same reuse principle covered for Terraform modules throughout this blog, applied here to an ML training pipeline.

Triggering pipelines automatically

from sagemaker.workflow.triggers import PipelineSchedule
 
pipeline.put_triggers(
    triggers=[PipelineSchedule(schedule_expression="cron(0 2 * * ? *)")],
)

A pipeline scheduled to run automatically — nightly here — retrains a model against fresh data on a defined cadence, without a person needing to remember to kick off the process manually. This is exactly the "rebuild and rescan on a schedule, independent of whether anyone remembered to trigger it" principle recommended for Docker base images earlier in this blog, applied here to model retraining specifically — genuinely important for a model whose accuracy can quietly degrade over time as the real-world data distribution it's predicting against shifts.

Triggering a pipeline from an event, not just a schedule

Beyond a fixed cron schedule, pipelines can also be triggered by an EventBridge rule reacting to a specific event — new data landing in S3, for instance, rather than waiting for the next scheduled run:

import boto3
 
events_client = boto3.client("events")
events_client.put_rule(
    Name="new-training-data-arrived",
    EventPattern=json.dumps({
        "source": ["aws.s3"],
        "detail-type": ["Object Created"],
        "detail": {"bucket": {"name": ["novuspark-training-data"]}},
    }),
)

Event-driven triggering is worth reaching for specifically when new training data arrives unpredictably rather than on a fixed cadence — a data source that's updated irregularly by an upstream system benefits more from "retrain when new data actually shows up" than from a fixed nightly schedule that might run against stale data most nights and miss a genuinely important update for hours on a schedule-driven cadence alone.

Model Registry: versioning models the way you'd version any other artifact

client = boto3.client("sagemaker")
response = client.list_model_packages(ModelPackageGroupName="churn-prediction-models")
for pkg in response["ModelPackageSummaryList"]:
    print(pkg["ModelPackageArn"], pkg["ModelApprovalStatus"])

Every model that passes the quality gate gets registered as a versioned model package — approved for deployment, or left pending an explicit human approval step, the same manual-approval-gate pattern covered for GitHub Environments and LangGraph's human-in-the-loop interrupts elsewhere in this blog. This gives an organization an actual audit trail: which specific model version is currently deployed, when it was trained, what data and evaluation metrics produced it — rather than a deployed model with no clear record of how it got there.

Approving a model and deploying it as part of the same pipeline

The Model Registry's approval status can itself trigger the next stage of automation — a separate deployment pipeline listening for an approval event:

client.update_model_package(
    ModelPackageArn="arn:aws:sagemaker:eu-west-2:008971632408:model-package/churn-prediction-models/7",
    ModelApprovalStatus="Approved",
)

# A second, separate pipeline (or a Lambda triggered by the approval event)
def deploy_approved_model(event, context):
    model_package_arn = event["detail"]["ModelPackageArn"]
    model = ModelPackage(model_package_arn=model_package_arn, role=role)
    model.deploy(initial_instance_count=1, instance_type="ml.m5.large", endpoint_name="churn-prediction-endpoint")

Separating the training pipeline (produces a candidate model, registers it pending approval) from the deployment trigger (fires once a human or an automated policy approves a specific registered version) mirrors the same separation covered for GitHub Actions environments requiring manual approval before a production deploy proceeds: automation handles everything up to the point a genuine decision is needed, and a deliberate approval step — not an assumption — is what actually authorizes the next stage.

What to actually remember from this post

• A Pipeline replaces manually-run notebook cells with an explicit, automated, code-defined sequence of steps — the same fix GitHub Actions provides for a manual deploy process, applied to the ML lifecycle specifically.
• Conditional steps act as quality gates, registering a model only if it meets a defined performance bar — comparing against the currently deployed model's own metrics is a meaningfully stronger check than a fixed threshold alone.
• Parameterize a pipeline definition the same way you'd parameterize a Terraform module — one definition, reusable across environments and datasets, not duplicated per use case.
• Scheduled and event-driven pipeline runs both keep a model current against a shifting real-world data distribution — choose based on whether new data arrives on a predictable cadence or unpredictably.
• The Model Registry provides a genuine audit trail of every trained model version, its approval status, and what produced it — and that approval status can itself trigger a separate, deliberate deployment step.

Next in the series: Cost Optimization on Amazon SageMaker, the final post — covering where SageMaker costs actually accumulate, and the concrete levers for controlling them.