10.1 SageMaker Pipelines Core Architecture

AWS Certified Machine Learning Engineer — Associate MLA-C01

Track your progress as you read

Finished sections get a check mark.

1 Chapter 1: Introduction & Exam Strategy

1.1 AWS MLA-C01 Blueprint, Format & Strategy 1.2 The Machine Learning Engineering Lifecycle on AWS

2 Domain 1: Data Ingestion & Storage Architecture for ML

2.1 Ingesting Batch and Streaming Data for ML 2.2 High-Performance Storage for Training and Inference 2.3 Data Cataloging, Governance, and Querying

3 Domain 1: Feature Engineering & Data Transformation

3.1 Distributed Feature Engineering with AWS Glue & DataBrew 3.2 Interactive Data Preparation with SageMaker Data Wrangler 3.3 Feature Management with SageMaker Feature Store

4 Domain 1: Data Quality, Integrity & Ground Truth Labeling

4.1 Data Validation, Schema Drift, and Quality Rules 4.2 Handling Missing Values, Outliers & Class Imbalance 4.3 Data Labeling with SageMaker Ground Truth & Ground Truth Plus

5 Domain 2: Model Selection & Framework Architecture

5.1 SageMaker Built-in Algorithms & Selection Criteria 5.2 Custom Models with Script Mode and BYOC 5.3 Foundation Models, JumpStart, and Amazon Bedrock

6 Domain 2: SageMaker Model Training & Hyperparameter Optimization

6.1 SageMaker Training Jobs & Cost Optimization 6.2 Distributed Training Strategies & Architectures 6.3 Hyperparameter Optimization with Automatic Model Tuning (AMT)

7 Domain 2: Model Evaluation, Tracking & Model Registry

7.1 ML Evaluation Metrics & Validation Strategies 7.2 Experiment Tracking with SageMaker Experiments 7.3 Model Governance with SageMaker Model Registry

8 Domain 3: SageMaker Inference Options & Endpoint Architecture

8.1 Real-Time, Serverless, and Asynchronous Inference 8.2 High-Throughput Offline Inference with Batch Transform 8.3 Multi-Model Endpoints (MME) & Multi-Container Endpoints (MCE)

9 Domain 3: Deployment Strategies & Infrastructure Scaling

9.1 Production Deployment Patterns & Traffic Shifting 9.2 Endpoint Auto-Scaling & Hardware Acceleration

10 Domain 3: MLOps Pipelines & Workflow Orchestration

10.1 SageMaker Pipelines Core Architecture 10.2 Advanced Pipeline Patterns & Integration 10.3 End-to-End MLOps CI/CD & Automated Retraining

11 Domain 4: Production Model Monitoring with SageMaker Model Monitor

11.1 Data Quality & Model Quality Monitoring 11.2 Model Bias & Model Explainability Drift 11.3 Operational Observability with Amazon CloudWatch

12 Domain 4: Responsible AI, Bias Detection & Governance

12.1 Pre-Training and Post-Training Bias with SageMaker Clarify 12.2 Model Documentation & Auditing with Model Cards & Dashboard 12.3 Safety, Guardrails & Generative AI Governance

13 Domain 4: ML Security, Network Isolation & Cost Optimization

13.1 IAM Roles and Least Privilege for ML 13.2 Network Isolation, VPC Endpoints & Data Encryption 13.3 ML Cost Optimization & Resource Governance

10.1 SageMaker Pipelines Core Architecture

Key Takeaways Explain This Topic AI

Amazon SageMaker Pipelines is a purpose-built, serverless ML workflow orchestration service defined declaratively via the SageMaker Python SDK ( sagemaker.workflow ) that executes Directed Acyclic Graphs (DAGs) without managing underlying infrastructure.

Pipeline parameters ( ParameterString , ParameterInteger , ParameterFloat ) allow dynamic runtime value injection and step overrides during StartPipelineExecution without altering the underlying DAG definition.

Fundamental pipeline step types include ProcessingStep (data transformation and feature extraction), TrainingStep (model training with Estimators), TuningStep (hyperparameter optimization with HyperparameterTuner ), TransformStep (batch scoring), and ConditionStep (conditional branching).

DAG execution dependencies are resolved automatically through property references (e.g., passing step_train.properties.ModelArtifacts.S3ModelArtifacts into an evaluation step) or explicitly using the depends_on parameter.

Step Caching via CacheConfig(enable_caching=True, expire_after="p30d") generates cryptographic hash keys from step attributes, container images, code hashes, and inputs to skip redundant, expensive compute operations when upstream inputs remain unchanged.

Last updated: August 2026

SageMaker Pipelines Core Architecture

Modern enterprise machine learning demands end-to-end repeatability, automated lineage tracking, deterministic execution, and seamless integration between data engineering and model training. Ad-hoc Jupyter notebook execution and fragmented bash scripts fail to provide the governance, auditability, and scalability required for production systems. Amazon SageMaker Pipelines is AWS's purpose-built, serverless Machine Learning Operations (MLOps) workflow engine designed specifically to compose, orchestrate, and automate machine learning lifecycles.

For the AWS Certified Machine Learning Engineer — Associate (MLA-C01) exam, you must understand how SageMaker Pipelines constructs Directed Acyclic Graphs (DAGs), how to declare and inject runtime parameters, how fundamental step types interact, how dependencies are formed, and how step caching optimizes compute costs and iteration speeds.

1. Declarative DAG Architecture & Execution Model

SageMaker Pipelines provides a declarative Python SDK ( sagemaker.workflow ) that allows data scientists and ML engineers to define complete workflows in code. When a pipeline is created or updated, SageMaker compiles the Python step definitions into a structured JSON pipeline definition graph that represents a Directed Acyclic Graph (DAG) . +---------------------------------------------------------------------------------------------------+

• | SAGEMAKER PIPELINES DECLARATIVE DAG ARCHITECTURE |
• | |
• | [SageMaker Python SDK Code] ---------> pipeline.upsert() ---------> [Compiled JSON DAG] |
• | - Step definitions (Control Plane Registration) - Nodes: Step Definitions |
• | - Parameter declarations - Edges: Data/Exec Links |
• | - Data / Execution dependencies |
• | |
• | pipeline.start() |
• | | |
• | v |
• | [Serverless Managed Execution Engine] |
• | | |
• | +-------------------------------+-------------------------------+ |
• | | | | |
• | v v v |
• | [Processing Step] [Training Step] [Condition Step] |
• | - Auto-provisions EC2 - Auto-provisions EC2 - Evaluates metric logic |
• | - Executes container - Trains model estimator - Routes DAG execution |
• | - Teardown on finish - Teardown on finish - Zero compute cost |
• +---------------------------------------------------------------------------------------------------+

Key Architectural Tenets:

Serverless Orchestration: You do not manage, provision, or patch an orchestration cluster (unlike Apache Airflow or Kubeflow). SageMaker manages the pipeline control plane at zero standby cost. You only pay for the underlying compute instances provisioned during individual step executions (e.g., Training or Processing instances).

Declarative Specification: Pipelines are defined using high-level Python objects. Calling pipeline.definition() outputs the underlying Amazon States Language / JSON representation used by the service.

Native SageMaker Integration: Built-in hooks automatically capture metadata into SageMaker Lineage Tracking , track experiments via SageMaker Experiments , package artifacts into SageMaker Model Registry , and monitor security via AWS IAM and AWS KMS .

2. Pipeline Parameters & Dynamic Execution Variables

To make pipeline DAGs reusable across staging and production environments, different dataset splits, or varying compute configurations, SageMaker Pipelines provides Pipeline Parameters and Execution Variables . +---------------------------------------------------------------------------------------------------+

• | PIPELINE RUNTIME PARAMETER INJECTION |
• | |
• | Pipeline Definition (Static Template): |
• | - model_approval_status = ParameterString(default_value="PendingManualApproval") |
• | - training_instance_type = ParameterString(default_value="ml.m5.xlarge") |
• | - batch_size = ParameterInteger(default_value=64) |
• | - learning_rate = ParameterFloat(default_value=0.001) |
• | |
• | | |
• | Execution Override at Runtime |
• | (StartPipelineExecution API) |
• | | |
• | v |
• | Runtime Execution Instance: |
• | - pipeline.start(parameters={"training_instance_type": "ml.p3.2xlarge", "batch_size": 128}) |
• +---------------------------------------------------------------------------------------------------+
• Parameter Primitive Types

SageMaker provides three parameter classes within sagemaker.workflow.parameters :

ParameterString : Strings such as S3 URIs, IAM role ARNs, Model Package Group names, or instance types.

ParameterInteger : Integers for epoch counts, batch sizes, worker counts, or maximum depth.

ParameterFloat : Floating point values for learning rates, regularization constants, or condition thresholds. from sagemaker.workflow.parameters import (

• ParameterString,
• ParameterInteger,
• ParameterFloat
• )

• # Defining pipeline parameters with robust defaults
• input_data_uri = ParameterString(
• name="InputDataUrl",
• default_value="s3://my-mlops-bucket/data/raw/dataset.csv"
• )
• training_instance = ParameterString(
• name="TrainingInstanceType",
• default_value="ml.c6i.2xlarge"
• )
• learning_rate = ParameterFloat(
• name="LearningRate",
• default_value=0.01
• )
• max_depth = ParameterInteger(
• name="MaxDepth",
• default_value=6
• )
• model_approval = ParameterString(
• name="ModelApprovalStatus",
• default_value="PendingManualApproval"
• )
• Dynamic Execution Variables ( ExecutionVariables )

In addition to custom parameters, SageMaker injects dynamic context variables at runtime accessible via sagemaker.workflow.execution_variables.ExecutionVariables :

ExecutionVariables.PIPELINE_EXECUTION_ID : Unique UUID for the execution run (useful for tagging S3 output paths).

ExecutionVariables.PIPELINE_EXECUTION_ARN : Full ARN of the active execution.

ExecutionVariables.START_DATETIME : Execution start timestamp in ISO 8601 format.

3. Fundamental Pipeline Step Types

SageMaker Pipelines includes specialized step classes in sagemaker.workflow.steps . Each step abstracts an AWS infrastructure task, runs inside a managed container, and exposes strongly typed properties to downstream steps. +---------------------------------------------------------------------------------------------------+

• | CORE PIPELINE STEP TYPES SPECTRUM |
• | |
• | +-------------------+ +-------------------+ +-------------------+ +----------------+ |
• | | ProcessingStep | | TrainingStep | | TuningStep | | ConditionStep | |
• | +-------------------+ +-------------------+ +-------------------+ +----------------+ |
• | | ScriptProcessor | | SageMaker | | Hyperparameter | | Compares | |
• | | DataWrangler | | Estimator | | Tuner | | metrics | |
• | | PySparkProcessor | | Model artifact | | Multi-job HPO | | (if_steps / | |
• | | Feature scaling | | output S3 tar.gz | | Best model search | | else_steps) | |
• | +-------------------+ +-------------------+ +-------------------+ +----------------+ |
• +---------------------------------------------------------------------------------------------------+
• 1. ProcessingStep (Data Preprocessing & Feature Engineering)

Wraps SageMaker Processing jobs using processors such as ScriptProcessor , SKLearnProcessor , PySparkProcessor , or DataWranglerProcessor .

Inputs: Configured via ProcessingInput to download data from Amazon S3 into the container filesystem ( /opt/ml/processing/input ).

Outputs: Configured via ProcessingOutput to upload transformed data from /opt/ml/processing/output back to Amazon S3.

Properties: Exposes output paths via step_process.properties.ProcessingOutputConfig.Outputs['train_data'].S3Output.S3Uri . from sagemaker.sklearn.processing import SKLearnProcessor

• from sagemaker.processing import ProcessingInput, ProcessingOutput
• from sagemaker.workflow.steps import ProcessingStep

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

• step_process = ProcessingStep(
• name="PreprocessFeatureEngineering",
• processor=sklearn_processor,
• inputs=[
• ProcessingInput(source=input_data_uri, destination="/opt/ml/processing/input")
• ],
• outputs=[
• ProcessingOutput(output_name="train", source="/opt/ml/processing/train"),
• ProcessingOutput(output_name="test", source="/opt/ml/processing/test")
• ],
• code="src/preprocessing.py"
• )
• 2. TrainingStep (Model Training)

Wraps a standard SageMaker Estimator (e.g., XGBoost, PyTorch, TensorFlow, or custom container).

Inputs: Configured using TrainingInput referencing S3 paths generated by upstream ProcessingStep outputs.

Outputs: Produces trained model weights packaged into model.tar.gz stored in S3.

Properties: Downstream steps reference model artifacts via step_train.properties.ModelArtifacts.S3ModelArtifacts . from sagemaker.estimator import Estimator

• from sagemaker.inputs import TrainingInput
• from sagemaker.workflow.steps import TrainingStep

• xgb_estimator = Estimator(
• image_uri=image_uri,
• instance_type=training_instance,
• instance_count=1,
• hyperparameters={"max_depth": max_depth, "eta": learning_rate},
• role=role
• )

• step_train = TrainingStep(
• name="TrainCustomerChurnModel",
• estimator=xgb_estimator,
• inputs={
• "train": TrainingInput(
• s3_data=step_process.properties.ProcessingOutputConfig.Outputs["train"].S3Output.S3Uri,
• content_type="text/csv"
• )
• }
• )
• 3. TuningStep (Hyperparameter Optimization)

Wraps a SageMaker HyperparameterTuner to run distributed hyperparameter optimization within the pipeline DAG.

Automatically orchestrates multiple training trials searching continuous, integer, or categorical hyperparameter ranges.

Exposes helper functions such as step_tune.get_top_model_s3_uri(top_k=0, s3_bucket=bucket_name) to automatically retrieve the best-performing model artifact for downstream evaluation or registration.

4. TransformStep (Batch Inference / Offline Scoring)

Wraps a SageMaker Transformer to execute offline batch scoring over large datasets stored in S3 without deploying a real-time HTTPS endpoint.

Useful for scoring entire customer databases overnight or generating offline embeddings.

5. ConditionStep (Branching Logic)

Implements conditional branching based on logical comparisons.

Compares an evaluation metric (e.g., test set ROC-AUC extracted from a JSON evaluation report) against a predetermined threshold.

Defines two branch paths: if_steps (executed when condition evaluates to True ) and else_steps (executed when False ).

Zero Compute Cost: Evaluates instantly in the SageMaker Pipelines control plane without spinning up an EC2 instance. from sagemaker.workflow.conditions import ConditionGreaterThanOrEqualTo

• from sagemaker.workflow.condition_step import ConditionStep
• from sagemaker.workflow.properties import PropertyFile

• # PropertyFile parses JSON evaluation output from a ProcessingStep
• evaluation_report = PropertyFile(
• name="EvaluationReport",
• output_name="evaluation",
• path="evaluation.json"
• )

from sagemaker.workflow.functions import JsonGet

• cond_gte = ConditionGreaterThanOrEqualTo(
• left=JsonGet(
• step_name=step_eval.name,
• property_file=evaluation_report,
• json_path="metrics.auc.value" # Extracts the AUC value from evaluation.json
• ),
• right=0.80 # Minimum 80% AUC requirement
• )

• step_cond = ConditionStep(
• name="CheckModelQualityGate",
• conditions=[cond_gte],
• if_steps=[step_register],
• else_steps=[step_fail]
• )
• 4. Establishing DAG Dependencies (Implicit vs. Explicit)

In SageMaker Pipelines, you rarely need to manually construct edge lists. The DAG resolves dependencies automatically: +---------------------------------------------------------------------------------------------------+

• | DAG DEPENDENCY RESOLUTION FLOW |
• | |
• | [Step 1: ProcessingStep] |
• | | |
• | | (Implicit Data Dependency: step_process.properties.Outputs['train']) |
• | v |
• | [Step 2: TrainingStep] |
• | | |
• | | (Implicit Data Dependency: step_train.properties.ModelArtifacts) |
• | v |
• | [Step 3: Evaluation ProcessingStep] |
• | | |
• | | (Control Flow Dependency: condition evaluation) |
• | v |
• | [Step 4: ConditionStep] |
• | | |
• | +-------------------------------+-------------------------------+ |
• | | (if True) | | (if False) |
• | v v v |
• | [Step 5a: RegisterModel] | [Step 5b: FailStep] |
• | | |
• | [Step 6: Custom Audit Step] <-------------+ (Explicit Dependency: depends_on=[step_process]) |
• +---------------------------------------------------------------------------------------------------+

Implicit Data Dependencies (Property References): When a step consumes an output property of an upstream step (e.g., step_train consuming step_process.properties.ProcessingOutputConfig... ), SageMaker automatically infers that step_process must complete successfully before step_train can start.

Explicit Execution Dependencies ( depends_on ): If Step B does not consume data outputs from Step A, but must only run after Step A completes (e.g., a data cleanup task or an external audit log), you explicitly pass depends_on=[step_a] to Step B.

5. Step Caching with CacheConfig

Data preprocessing, feature engineering, and hyperparameter tuning jobs can take hours and cost hundreds of dollars. During pipeline development and iterative debugging, rerunning identical compute steps when input data and parameters have not changed is wasteful. +---------------------------------------------------------------------------------------------------+

• | SAGEMAKER STEP CACHING MECHANISM |
• | |
• | Step Invocation: |
• | 1. Pipeline generates Cache Key Hash: |
• | Hash = SHA256( Code_URI + Container_Image_Digest + Input_S3_Etag + Parameters ) |
• | |
• | 2. Lookup Cache Store: |
• | +-----------------------------------------------------------------------+ |
• | | Does valid cache entry exist within 'expire_after' window (e.g., p30d)?| |
• | +-----------------------------------------------------------------------+ |
• | | | |
• | v (YES: CACHE HIT) v (NO: CACHE MISS) |
• | [Instant Step Completion] [Provision EC2 Compute Instance] |
• | - Time: ~1 second - Time: 15-60 minutes |
• | - Compute Cost: $0.00 - Compute Cost: Standard hourly rate |
• | - Reuses previous S3 outputs - Writes new S3 outputs & updates cache |
• +---------------------------------------------------------------------------------------------------+
• Configuring Step Caching: from sagemaker.workflow.steps import CacheConfig

• # Configure step caching with a 30-day expiration window
• cache_config = CacheConfig(
• enable_caching=True,
• expire_after="p30d" # ISO 8601 duration format (30 days)
• )

• step_process = ProcessingStep(
• name="CachedPreprocessingStep",
• processor=sklearn_processor,
• inputs=[ProcessingInput(source=input_data_uri, destination="/opt/ml/processing/input")],
• outputs=[ProcessingOutput(output_name="train", source="/opt/ml/processing/train")],
• code="src/preprocessing.py",
• cache_config=cache_config # Attach caching configuration
• )

Cache Key Composition & Invalidation:

SageMaker automatically computes a deterministic hash key based on:

• Pipeline Name and Step Name
• Container Image URI and Digest
• Step Code / Script Content (any code change invalidates cache)
• Input S3 URIs and Object S3 ETag/Checksums (new data files invalidate cache)
• Input Parameters and Hyperparameters

Tip

Exam Rule for Step Caching: Step caching uses ISO 8601 duration strings (e.g., p30d for 30 days, pt12h for 12 hours, p1y for 1 year). If any input parameter, S3 object hash, or script changes, SageMaker automatically invalidates the cache and executes a fresh run.

Loading diagram... SageMaker Pipelines Core Step Architecture & DAG Flow