Module 9 — SageMaker Pipelines and the Model Registry

Modules 4 to 8 launched jobs by hand. In production, that is not how it works: training, evaluation and deployment run together, on a schedule or on a code change, and the model that ends up in front of users is the one whose evaluation crossed a metric threshold — not the last one that trained. SageMaker Pipelines encodes that logic, and the Model Registry stores the artifacts it produces.

What a pipeline actually is

A Pipeline is a directed acyclic graph of steps, defined in Python and compiled to a JSON definition that AWS executes. The four steps that matter for our case: ProcessingStep, TrainingStep, ProcessingStep (evaluation), and ConditionStep gating a ModelStep or RegisterModelStep.

The example below chains them for the churn model. It compiles in one file and requires no Airflow, no cron, no CI plumbing.

from sagemaker.workflow.pipeline import Pipeline
from sagemaker.workflow.parameters import ParameterString, ParameterFloat
from sagemaker.workflow.steps import ProcessingStep, TrainingStep
from sagemaker.workflow.condition_step import ConditionStep
from sagemaker.workflow.conditions import ConditionGreaterThanOrEqualTo
from sagemaker.workflow.functions import JsonGet
from sagemaker.workflow.model_step import ModelStep
from sagemaker.workflow.properties import PropertyFile
from sagemaker.sklearn.processing import SKLearnProcessor
from sagemaker.processing import ProcessingInput, ProcessingOutput

# --- Parameters exposed at execution time ---
input_data = ParameterString(name="InputData", default_value=f"s3://{bucket}/churn/raw/")
auc_threshold = ParameterFloat(name="AucThreshold", default_value=0.80)

Parameters at the pipeline level are what you change per run without editing the definition — the input date, a metric threshold, an approval requirement. Anything you hard-code becomes an obstacle later.

The processing step

processor = SKLearnProcessor(
    framework_version="1.2-1", role=role,
    instance_type="ml.m5.xlarge", instance_count=1,
)

process_step = ProcessingStep(
    name="Preprocess",
    processor=processor,
    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"),
    ],
    code="preprocess.py",
)

The processing step runs an arbitrary Python script on a managed container. Its outputs are referenced by name in downstream steps, decoupling storage paths from the graph.

Training step, wired to the previous outputs

train_step = TrainingStep(
    name="Train",
    estimator=sk,
    inputs={
        "train": TrainingInput(
            s3_data=process_step.properties.ProcessingOutputConfig.Outputs["train"].S3Output.S3Uri,
            content_type="application/x-parquet",
        ),
        "validation": TrainingInput(
            s3_data=process_step.properties.ProcessingOutputConfig.Outputs["validation"].S3Output.S3Uri,
            content_type="application/x-parquet",
        ),
    },
)

That process_step.properties... expression is the lazy reference that turns a set of independent jobs into a graph. At definition time the S3 path does not exist yet; at execution time SageMaker resolves it after the processing step completes.

Evaluation and condition

The evaluation writes a JSON file with the metric; the condition step reads that JSON via JsonGet and gates the rest of the graph on the value.

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

eval_step = ProcessingStep(
    name="Evaluate",
    processor=processor,
    inputs=[
        ProcessingInput(source=train_step.properties.ModelArtifacts.S3ModelArtifacts,
                        destination="/opt/ml/processing/model"),
        ProcessingInput(source=process_step.properties.ProcessingOutputConfig.Outputs["test"].S3Output.S3Uri,
                        destination="/opt/ml/processing/test"),
    ],
    outputs=[ProcessingOutput(output_name="evaluation", source="/opt/ml/processing/evaluation")],
    code="evaluate.py",
    property_files=[evaluation_report],
)

condition = ConditionGreaterThanOrEqualTo(
    left=JsonGet(step_name=eval_step.name, property_file=evaluation_report, json_path="metrics.auc.value"),
    right=auc_threshold,
)

The evaluate.py script writes to /opt/ml/processing/evaluation/evaluation.json a document of the form {"metrics": {"auc": {"value": 0.847}}}. JsonGet extracts the leaf; the condition compares it to the parameter; the graph continues only if the comparison is true.

Register the model — with an approval status

Above the threshold, register the model with PendingManualApproval so a human confirms before deployment.

from sagemaker.workflow.model_step import ModelStep
from sagemaker.model_metrics import ModelMetrics, MetricsSource

model_metrics = ModelMetrics(model_statistics=MetricsSource(
    s3_uri=eval_step.properties.ProcessingOutputConfig.Outputs["evaluation"].S3Output.S3Uri + "/evaluation.json",
    content_type="application/json",
))

register_step = ModelStep(
    name="Register",
    step_args=model.register(
        content_types=["application/json"], response_types=["application/json"],
        inference_instances=["ml.m5.large"], transform_instances=["ml.m5.xlarge"],
        model_package_group_name="churn",
        approval_status="PendingManualApproval",
        model_metrics=model_metrics,
    ),
)

gate = ConditionStep(
    name="AucAboveThreshold",
    conditions=[condition],
    if_steps=[register_step],
    else_steps=[],
)

pipeline = Pipeline(
    name="churn-training",
    parameters=[input_data, auc_threshold],
    steps=[process_step, train_step, eval_step, gate],
)
pipeline.upsert(role_arn=role)

pipeline.upsert compiles the definition and pushes it. pipeline.start(parameters={"AucThreshold": 0.82}) runs it, and every run appears in Studio with the DAG, the logs and the S3 artifacts of each step.

The Model Registry, in one paragraph

A model package group ("churn") holds versions. Each RegisterModel adds a version, with its metrics and its status: PendingManualApproval, Approved, Rejected. The endpoint of module 7 can be configured to pick "the latest Approved version of churn", which is what makes a controlled rollout automatable. Manual approval can be automated on stricter conditions than the AUC threshold — for example a fairness metric — and rejection leaves an audit trail of the models that were considered and not deployed. This closes the loop with course 20's registry: same idea, native tooling.

A condition step does not stop the pipeline

When the metric is below the threshold, the else_steps list runs. An empty else_steps just ends the branch, but the pipeline itself succeeds — no error, no page. If a subthreshold model should raise an alarm, add a FailStep in else_steps; otherwise a degraded model slips by unnoticed.

Summary

• A Pipeline is a DAG of ProcessingStep, TrainingStep, ConditionStep and ModelStep; parameters expose what varies per run.
• Steps reference each other's outputs by lazy properties; SageMaker resolves them at execution time.
• The evaluation step writes a JSON, JsonGet reads a value, ConditionStep gates the register step on it.
• The Model Registry stores versions with an Approved / Rejected status; a condition step's else_steps needs a FailStep or the pipeline is silently green.

Next module: watch the deployed model with Model Monitor and cap the invoice.