Building Your First SageMaker Pipeline

Building Your First SageMaker Pipeline | CodeSignal Learn

Introduction & Lesson Overview

Welcome to an exciting new chapter in your machine learning journey with AWS SageMaker! By now, you've mastered the individual components of machine learning workflows — you know how to preprocess datasets, train models in SageMaker, and deploy them for real-world use. These are powerful skills, but there's one more crucial piece to complete your ML toolkit: automation . In this lesson, you'll learn how to connect all these individual steps into a seamless, automated workflow using SageMaker Pipelines . Think of it as moving from cooking individual dishes to orchestrating an entire restaurant kitchen — each step needs to happen in the right order, with the right inputs, and you want it all to run smoothly without manual intervention. By the end of this lesson, you'll have built your first SageMaker Pipeline with a data preprocessing step. This foundation will prepare you for the more complex pipelines we'll build in upcoming lessons, where we'll add training, evaluation, and conditional model registration steps. You'll understand not just how to write the code, but why each component matters and how they work together to create robust, production-ready ML workflows.

What are SageMaker Pipelines?

SageMaker Pipelines is AWS's solution for creating and managing machine learning workflows. A pipeline is essentially a series of connected steps that execute in a specific order, where each step takes inputs, processes them, and produces outputs that become inputs for the next step. Just like an assembly line in manufacturing, each step takes inputs, processes them, and produces outputs that become inputs for the next step. The real power of pipelines becomes clear when you consider what happens without them. Imagine you're working on a machine learning project where you need to preprocess data, train a model, evaluate its performance, and then deploy it only if it meets certain quality criteria. Without automation, you'd need to manually run each step, wait for it to complete, check the results, and then decide whether to proceed to the next step. This process is time-consuming, error-prone, and doesn't scale well when you need to retrain models regularly or work with multiple datasets. SageMaker Pipelines solves these challenges by providing several key benefits: Reproducibility — your pipeline will execute the same way every time, eliminating the "it worked on my machine" problem Scalability — you can easily run the same pipeline on different datasets or with different parameters Monitoring and tracking — you can see exactly what happened at each step, making debugging and optimization much easier Collaboration — team members can understand and modify the workflow without needing to decipher scattered scripts and manual processes These benefits transform machine learning from a collection of manual, error-prone tasks into a reliable, automated system that can scale with your business needs.

Join the 1M+ learners on CodeSignal

Be a part of our community of 1M+ users who develop and demonstrate their skills on CodeSignal

Adapting Preprocessing Script for SageMaker

To build a pipeline, we need to start with individual steps. Our first step will be data preprocessing — the same preprocessing you've done locally in previous courses, but now adapted to run in SageMaker's managed environment. We'll create a separate file called data_processing.py that contains our preprocessing logic. The preprocessing logic itself is identical to what you've done before — we're still capping outliers, creating new features, and splitting the data. The key changes are in how we handle file paths to work within SageMaker's processing environment: Pythonimport os import pandas as pd from sklearn.model_selection import train_test_split if __name__ == "__main__": # Read input data from SageMaker's input directory input_data_path = "/opt/ml/processing/input/california_housing.csv" df = pd.read_csv(input_data_path) print(f"Processing {len(df)} samples...") # Apply the same preprocessing steps you've done before: # 1. Cap outliers at 95th percentiles (excluding geographic coordinates) # 2. Create RoomsPerHousehold feature # 3. Select relevant features for modeling # ... (preprocessing logic here) ... # Split the data into training and testing sets (80% train, 20% test) X_train, X_test, y_train, y_test = train_test_split( X, y, test_size=0.2, random_state=42 ) # Create output directories for SageMaker os.makedirs("/opt/ml/processing/train", exist_ok=True) os.makedirs("/opt/ml/processing/test", exist_ok=True) # Combine features and target for saving train_data = pd.concat([X_train, y_train], axis=1) test_data = pd.concat([X_test, y_test], axis=1) # Save to SageMaker's expected output directories train_data.to_csv("/opt/ml/processing/train/train.csv", index=False) test_data.to_csv("/opt/ml/processing/test/test.csv", index=False) print("Data processing completed!") print(f"Train samples: {len(X_train)}") print(f"Test samples: {len(X_test)}")import os import pandas as pd from sklearn.model_selection import train_test_split if __name__ == "__main__": # Read input data from SageMaker's input directory input_data_path = "/opt/ml/processing/input/california_housing.csv" df = pd.read_csv(input_data_path) print(f"Processing {len(df)} samples...") # Apply the same preprocessing steps you've done before: # 1. Cap outliers at 95th percentiles (excluding geographic coordinates) # 2. Create RoomsPerHousehold feature # 3. Select relevant features for modeling # ... (preprocessing logic here) ... # Split the data into training and testing sets (80% train, 20% test) X_train, X_test, y_train, y_test = train_test_split( X, y, test_size=0.2, random_state=42 ) # Create output directories for SageMaker os.makedirs("/opt/ml/processing/train", exist_ok=True) os.makedirs("/opt/ml/processing/test", exist_ok=True) # Combine features and target for saving train_data = pd.concat([X_train, y_train], axis=1) test_data = pd.concat([X_test, y_test], axis=1) # Save to SageMaker's expected output directories train_data.to_csv("/opt/ml/processing/train/train.csv", index=False) test_data.to_csv("/opt/ml/processing/test/test.csv", index=False) print("Data processing completed!") print(f"Train samples: {len(X_train)}") print(f"Test samples: {len(X_test)}") The three critical changes that make this script SageMaker-compatible are: Input path: We read from /opt/ml/processing/input/ instead of a local file Output paths: We write to /opt/ml/processing/train/ and /opt/ml/processing/test/ directories Directory creation: We explicitly create output directories using os.makedirs() These specific paths (/opt/ml/processing/) are SageMaker conventions that allow the service to automatically handle data movement between S3 and your processing containers. When SageMaker runs this script, it will automatically mount your S3 data to the input directory and upload the results from the output directories back to S3.

Understanding SageMaker Sessions

Setting Up AWS Resources

Now that we understand sessions, we need to set up the AWS resources and permissions that our pipeline will use: Python# Retrieve your AWS account ID (used for constructing resource ARNs) account_id = sagemaker_session.boto_session.client('sts').get_caller_identity()['Account'] # Get the default S3 bucket for your SageMaker resources default_bucket = sagemaker_session.default_bucket() # Define the SageMaker execution role ARN SAGEMAKER_ROLE = f"arn:aws:iam::{account_id}:role/SageMakerDefaultExecution"# Retrieve your AWS account ID (used for constructing resource ARNs) account_id = sagemaker_session.boto_session.client('sts').get_caller_identity()['Account'] # Get the default S3 bucket for your SageMaker resources default_bucket = sagemaker_session.default_bucket() # Define the SageMaker execution role ARN SAGEMAKER_ROLE = f"arn:aws:iam::{account_id}:role/SageMakerDefaultExecution" These resources should be familiar from your previous SageMaker work — the execution role provides the necessary permissions for SageMaker to access your data and create resources, while the default bucket provides storage for your pipeline inputs and outputs.

Creating the Processing Environment

Building Our First Pipeline Step

With our processor configured, we can now create the actual processing step using ProcessingStep. This step defines what data goes in, what comes out, and what code runs in between. Before proceeding, note that we assume you have already uploaded your raw dataset (california_housing.csv) to your S3 default bucket at the path /datasets/california_housing.csv. This is necessary because the pipeline will read the input data directly from S3. Pythonfrom sagemaker.workflow.steps import ProcessingStep # Define the processing step with inputs, outputs, and the script to run processing_step = ProcessingStep( name="ProcessData", # Unique name for this step in the pipeline processor=processor, # The processor we defined above inputs=[ # Define where the raw data comes from (S3 location) sagemaker.processing.ProcessingInput( source=f"s3://{default_bucket}/datasets/california_housing.csv", destination="/opt/ml/processing/input" # Where data will be mounted in container ) ], outputs=[ # Define where processed training data will be saved sagemaker.processing.ProcessingOutput( output_name="train_data", # Reference name for this output source="/opt/ml/processing/train" # Container path where script saves data ), # Define where processed test data will be saved sagemaker.processing.ProcessingOutput( output_name="test_data", # Reference name for this output source="/opt/ml/processing/test" # Container path where script saves data ) ], code="data_processing.py" # The Python script that performs the processing )from sagemaker.workflow.steps import ProcessingStep # Define the processing step with inputs, outputs, and the script to run processing_step = ProcessingStep( name="ProcessData", # Unique name for this step in the pipeline processor=processor, # The processor we defined above inputs=[ # Define where the raw data comes from (S3 location) sagemaker.processing.ProcessingInput( source=f"s3://{default_bucket}/datasets/california_housing.csv", destination="/opt/ml/processing/input" # Where data will be mounted in container ) ], outputs=[ # Define where processed training data will be saved sagemaker.processing.ProcessingOutput( output_name="train_data", # Reference name for this output source="/opt/ml/processing/train" # Container path where script saves data ), # Define where processed test data will be saved sagemaker.processing.ProcessingOutput( output_name="test_data", # Reference name for this output source="/opt/ml/processing/test" # Container path where script saves data ) ], code="data_processing.py" # The Python script that performs the processing ) Notice how the paths in our ProcessingInput and ProcessingOutput definitions perfectly match the paths our script expects. The inputs parameter specifies where our raw data comes from (an S3 location) and where it will be mounted inside the processing container (/opt/ml/processing/input). The outputs parameter defines where our processed data will be saved, with separate outputs for training and test data. Each output has a name that we can reference later and a source path where our processing script will write the data. The code parameter points to the Python script that contains our preprocessing logic. This connection between the step definition and the script paths is crucial — it's what allows SageMaker to automatically handle all the data movement for you.

Creating the Pipeline

Executing the Pipeline

Monitoring Your Pipeline

Once your pipeline is running, you can monitor its progress and check its status. The execution object provides several methods for tracking the pipeline's progress: Python# Check the current status execution_details = execution.describe() # Display status print(f"Status: {execution_details['PipelineExecutionStatus']}")# Check the current status execution_details = execution.describe() # Display status print(f"Status: {execution_details['PipelineExecutionStatus']}") When you run this code, you'll see output similar to: textStatus: ExecutingStatus: Executing The status will initially show "Executing" and will change to "Succeeded" or "Failed" as the pipeline progresses through its steps. You can also wait for the pipeline to complete: Python# Wait for completion (optional) execution.wait()# Wait for completion (optional) execution.wait() When successful, your processed data will be automatically saved to S3 in your default bucket under a path that includes the pipeline execution ID. For example, if your default bucket is sagemaker-us-east-1-123456789012 and the pipeline execution ID is x1gc33lgj8v5, the training split produced by the ProcessData step will be saved at: texts3://sagemaker-us-east-1-123456789012/california-housing-preprocessing-pipeline/x1gc33lgj8v5/ProcessData/output/train_data/s3://sagemaker-us-east-1-123456789012/california-housing-preprocessing-pipeline/x1gc33lgj8v5/ProcessData/output/train_data/ The corresponding test split would be saved at: texts3://sagemaker-us-east-1-123456789012/california-housing-preprocessing-pipeline/x1gc33lgj8v5/ProcessData/output/test_data/s3://sagemaker-us-east-1-123456789012/california-housing-preprocessing-pipeline/x1gc33lgj8v5/ProcessData/output/test_data/ This hierarchical structure makes it easy to track which pipeline run produced which outputs and helps organize your data as you iterate on your pipeline design. In the next steps, you’ll be able to easily access these S3 output locations as inputs for downstream steps in your pipeline, such as model training and evaluation.

Building Your First SageMaker Pipeline

Introduction & Lesson Overview

What are SageMaker Pipelines?

Join the 1M+ learners on CodeSignal

Adapting Preprocessing Script for SageMaker

Understanding SageMaker Sessions

Setting Up AWS Resources

Creating the Processing Environment

Building Our First Pipeline Step

Creating the Pipeline

Executing the Pipeline

Monitoring Your Pipeline

Summary & Next Steps