2. Structured ML Pipeline

Duration2h30 x 2 AI Banned

Introduction

This practical session focuses on transforming exploratory notebook code into structured, production-ready machine learning pipelines. You’ll learn to convert the air quality analysis notebook you previously developed into a modular, testable, and maintainable Python package that follows software engineering best practices.

Through this hands-on workshop, you will master essential skills for transitioning from data science experimentation to production deployment:

• Code Architecture: Transform monolithic notebook cells into modular, reusable classes with clear responsibilities
• Software Engineering Practices: Apply separation of concerns and proper error handling (advanced concepts like dependency injection are introduced but not fully implemented)
• Test-Driven Development: Understand the critical role of automated testing in ML pipelines through comprehensive unit tests
• Production Readiness: Create command-line interfaces and structured workflows suitable for deployment
• Professional Standards: Learn industry best practices used by Machine Learning Engineers in production environments

The challenge is not just to make the code work, but to understand the underlying architecture and complete missing implementations. This deep understanding of code structure will be invaluable for your future internships, projects, and professional work where you’ll need to build upon and extend existing ML systems.

Prerequisites

Before starting this activity, you should already have completed:

• Environment Setup - Python environment with required packages installed
• Git & GitLab How-to - Git basics
• Setup a Data Science project - Git workflow and project structure
• Air Quality Data Analysis - The notebook that serves as the foundation for this structured implementation

Do not forget to also read the introductory lesson:

• Towards Production-Ready Machine Learning

This activity assumes you have a working air quality analysis notebook and are familiar with the complete data science workflow from data loading to model evaluation.

Understanding the Target Architecture

The air quality project follows a modular pipeline architecture that separates concerns while maintaining clear data flow:

scripts/
├── crypto_utils.py
├── run_pipeline.py
├── run_tests.py
src/
├── pipeline/
│   ├── data_processor.py
│   ├── feature_engineer.py
│   ├── model_trainer.py
│   └── evaluator.py
└── utils/
    ├── config.py
    └── utils.py

Core Components

• DataProcessor: Handles data loading, cleaning, and preprocessing operations
• FeatureEngineer: Manages feature extraction, encoding, and selection processes
• ModelTrainer: Handles model training, persistence, and prediction workflows
• Evaluator: Provides model evaluation metrics and cross-validation functionality

Pipeline Orchestration

• run_pipeline.py: Main execution script that orchestrates the complete ML pipeline by integrating all components in the correct sequence

Supporting Infrastructure

• config.py: Centralizes configuration parameters and constants
• utils.py: Provides general utility functions for plotting and data operations
• evaluation_utils.py: Offers advanced evaluation and visualization capabilities

Getting Started

Before implementing the structured architecture, you need to obtain the source code and set up your development environment.

The air quality project source code is available as a downloadable archive containing all necessary files and structure.

Environment Setup with uv

After downloading and extracting the project, set up your environment using uv with project.toml file.

1
2
3
4
5
6
7

# Navigate to the project directory
cd air_quality

# Use uv to lock and sync your enviromnent

# Verify installation with a quick test
uv run python scripts/run_tests.py --quick

Running Commands

Throughout this tutorial, use uv run to execute Python scripts:

1
2
3
4
5
6
7
8

# Run tests
uv run python scripts/run_tests.py

# Run the pipeline
uv run python scripts/run_pipeline.py --model linear

# Launch Jupyter notebook
uv run jupyter notebook

Initial Validation

Before starting your implementation, ensure the project structure is correct:

1
2
3
4
5
6
7
8
9

# Check project structure
ls -la src/pipeline/
ls -la src/utils/

# Verify test framework is working
uv run python scripts/run_tests.py --help

# Run initial tests (should show failures for unimplemented methods)
uv run python scripts/run_tests.py --verbose

You should see test failures for unimplemented TODO sections - this is expected and confirms the testing framework is working correctly.

Implementation Guide

You will transform your notebook code into the structured architecture by completing TODO sections in four main classes. Each method corresponds to specific sections of your original notebook, making the transition straightforward.

Implementation Overview

Complete the following components in order:

1. DataProcessor Implementation - Data loading, cleaning, and preprocessing operations
2. FeatureEngineer Implementation - Feature extraction, encoding, and selection processes
3. ModelTrainer Implementation - Model training, persistence, and prediction workflows
4. Evaluator Implementation - Model evaluation metrics and cross-validation functionality
5. Pipeline Integration - Integrate all steps into a Machine Learning pipeline

DataProcessor Implementation

The DataProcessor class handles all data-related operations, ensuring temporal and geographic consistency throughout the preprocessing pipeline.

Core Responsibilities:

• Loading training and test datasets
• Managing missing values with temporal awareness
• Creating geographic cross-validation folds
• Maintaining data integrity throughout preprocessing

Method: load_data()

Purpose: Load training and test datasets from CSV files.

Notebook Connection: This implements the data loading section where you read the CSV files into DataFrames.

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16

def load_data(self):
    """
    Load training and test datasets from CSV files.
    
    Load air quality datasets from configured file paths and store them
    as instance attributes for further processing.
    
    Parameters
    ----------
    None
    
    Returns
    -------
    tuple
        Tuple of (train_df, test_df) DataFrames
    """

Method: forward_back_fill()

Purpose: Handle missing values in time-series data while respecting geographic boundaries.

Notebook Connection: This implements the missing value handling section where you applied forward and backward fill within each city.

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23

def forward_back_fill(self, df, cols, group_col, date_col):
    """
    Fill missing values using forward and backward fill within each group.
    
    This method handles missing values in time-series data while respecting 
    geographic boundaries by processing each group independently.
    
    Parameters
    ----------
    df : pandas.DataFrame
        DataFrame to process
    cols : list
        List of columns to fill
    group_col : str
        Column to group by (e.g., 'city')
    date_col : str
        Date column for sorting
    
    Returns
    -------
    pandas.DataFrame
        DataFrame with missing values filled
    """

Method: handle_missing_values()

Purpose: Process both training and test datasets city by city to maintain consistency.

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19

def handle_missing_values(self, train_df, test_df):
    """
    Handle missing values for time-series data with geographic groupings.
    
    Process both training and test datasets city by city to maintain 
    consistency across geographic boundaries.
    
    Parameters
    ----------
    train_df : pandas.DataFrame
        Training dataset with missing values
    test_df : pandas.DataFrame
        Test dataset with missing values
    
    Returns
    -------
    tuple
        Tuple of (train_processed, test_processed) DataFrames
    """

Method: drop_high_missing_columns()

Purpose: Remove columns with excessive missing data to improve data quality and model performance.

Notebook Connection: This implements the data cleaning section where you identified and removed columns with too many missing values.

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19

def drop_high_missing_columns(self, train_df, test_df):
    """
    Drop columns with more than MISSING_THRESHOLD proportion of missing data.
    
    Analyzes missing data patterns in training set and applies the same
    column drops to both training and test sets to maintain consistency.
    
    Parameters
    ----------
    train_df : pandas.DataFrame
        Training dataset to analyze for missing data patterns
    test_df : pandas.DataFrame
        Test dataset to apply same column drops
    
    Returns
    -------
    tuple
        Tuple of (train_cleaned, test_cleaned) DataFrames with high-missing columns removed
    """

Method: create_geographic_folds()

Purpose: Create cross-validation folds that respect geographic boundaries.

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17

def create_geographic_folds(self, df):
    """
    Create cross-validation folds based on geographic grouping.
    
    Generate cross-validation folds that ensure complete cities are either 
    in training or validation, never both, to prevent data leakage.
    
    Parameters
    ----------
    df : pandas.DataFrame
        DataFrame containing data with geographic grouping column
    
    Returns
    -------
    pandas.DataFrame
        DataFrame with 'folds' column added
    """

FeatureEngineer Implementation

The FeatureEngineer class transforms raw data into meaningful features for machine learning models.

Core Responsibilities:

• Extracting temporal patterns (seasonality, trends)
• Creating geographic identifiers and location features
• Encoding categorical variables consistently
• Selecting optimal feature subsets using statistical and model-based methods

Method: extract_temporal_features()

Purpose: Extract time-based features that capture seasonal and cyclical patterns in air quality data.

Notebook Connection: This corresponds to the temporal feature engineering section where you extracted year, month, day, etc.

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19

def extract_temporal_features(self, df, date_col='date'):
    """
    Extract temporal features from a datetime column.
    
    Create time-based features that capture seasonal and cyclical patterns 
    in air quality data for improved model performance.
    
    Parameters
    ----------
    df : pandas.DataFrame
        DataFrame containing datetime column
    date_col : str, default='date'
        Name of the datetime column to extract features from
    
    Returns
    -------
    pandas.DataFrame
        DataFrame with additional temporal features
    """

Method: extract_geographic_features()

Purpose: Create location-based identifiers for different monitoring stations.

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17

def extract_geographic_features(self, df):
    """
    Create geographic features from latitude and longitude coordinates.
    
    Generate unique location identifiers by combining coordinate information
    to help models learn location-specific pollution patterns.
    
    Parameters
    ----------
    df : pandas.DataFrame
        DataFrame with site_latitude and site_longitude columns
    
    Returns
    -------
    pandas.DataFrame
        DataFrame with additional geographic features
    """

Method: encode_categorical_features()

Purpose: Convert categorical variables to numerical format while maintaining consistency between training and test sets.

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25

def encode_categorical_features(self, train_df, test_df):
    """
    Encode categorical features for machine learning models.
    
    Convert categorical variables to numerical format while ensuring 
    consistent encoding between training and test datasets.
    
    Parameters
    ----------
    train_df : pandas.DataFrame
        Training dataset with categorical features
    test_df : pandas.DataFrame
        Test dataset with categorical features
    
    Returns
    -------
    tuple
        Tuple containing:
        - train_encoded : pandas.DataFrame
            Encoded training DataFrame
        - test_encoded : pandas.DataFrame
            Encoded test DataFrame
        - feature_cols : list
            List of column names to use as model features
    """

Method: select_features_selectkbest()

Purpose: Select features using statistical significance testing.

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27

def select_features_selectkbest(self, X_train, y_train, k=15):
    """
    Select features using SelectKBest with f_regression scoring.
    
    Use statistical tests to identify features with the strongest linear 
    relationships to the target variable.
    
    Parameters
    ----------
    X_train : pandas.DataFrame or numpy.ndarray
        Training features
    y_train : pandas.Series or numpy.ndarray
        Training target values
    k : int, default=15
        Number of top features to select
    
    Returns
    -------
    tuple
        Tuple containing:
        - X_train_selected : numpy.ndarray
            Training features after selection
        - selected_features : numpy.ndarray
            Names of selected features
        - feature_scores : numpy.ndarray
            Scores for all features
    """

Method: select_features_rfe()

Purpose: Select features using Recursive Feature Elimination with a linear model.

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27

def select_features_rfe(self, X_train, y_train, n_features=15):
    """
    Select features using Recursive Feature Elimination.
    
    Use model-based recursive elimination to identify optimal feature 
    combinations that work well together.
    
    Parameters
    ----------
    X_train : pandas.DataFrame or numpy.ndarray
        Training features
    y_train : pandas.Series or numpy.ndarray
        Training target values
    n_features : int, default=15
        Number of features to select
    
    Returns
    -------
    tuple
        Tuple containing:
        - X_train_selected : numpy.ndarray
            Training features after selection
        - selected_features : numpy.ndarray
            Names of selected features
        - feature_rankings : numpy.ndarray
            Rankings for all features
    """

ModelTrainer Implementation

The ModelTrainer class handles model creation and training.

Core Responsibilities:

• Training models with proper configuration
• Managing model hyperparameters
• Creating model instances for different algorithms

Method: train_single_model()

Purpose: Train a single model instance on provided data.

Notebook Connection: This corresponds to the model training sections where you fit LinearRegression models.

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21

def train_single_model(self, model, X_train, y_train):
    """
    Train a single model on the provided data.
    
    Fit a machine learning model on training data and return the trained 
    instance for prediction or evaluation.
    
    Parameters
    ----------
    model : sklearn estimator
        Sklearn model instance to train
    X_train : pandas.DataFrame or numpy.ndarray
        Training features
    y_train : pandas.Series or numpy.ndarray
        Training target values
    
    Returns
    -------
    sklearn estimator
        Trained model instance
    """

Evaluator Implementation

The Evaluator class provides comprehensive model assessment and cross-validation functionality.

Core Responsibilities:

• Calculating standard regression metrics (RMSE, R², MSE)
• Performing cross-validation across geographic folds
• Comparing multiple models systematically
• Generating performance reports and summaries

Method: calculate_metrics()

Purpose: Calculate standard regression metrics for model evaluation.

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19

def calculate_metrics(self, y_true, y_pred):
    """
    Calculate regression metrics for model evaluation.
    
    Compute standard regression performance metrics to assess model 
    quality and prediction accuracy.
    
    Parameters
    ----------
    y_true : pandas.Series or numpy.ndarray
        True target values
    y_pred : pandas.Series or numpy.ndarray
        Predicted target values
    
    Returns
    -------
    dict
        Dictionary containing RMSE, MSE, and R² metrics
    """

Method: cross_validate_model()

Purpose: Perform cross-validation using GroupKFold to ensure geographic integrity and prevent data leakage.

Notebook Connection: This implements the model evaluation section where you performed cross-validation to assess model performance across different data splits.

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26

def cross_validate_model(self, model, X, y, groups=None):
    """
    Perform cross-validation using GroupKFold.
    
    This method uses GroupKFold to ensure entire cities are either in training 
    OR validation, never both, preventing data leakage.
    
    Parameters
    ----------
    model : sklearn estimator
        Scikit-learn model to evaluate
    X : pandas.DataFrame or numpy.ndarray
        Feature matrix
    y : pandas.Series or numpy.ndarray
        Target variable
    groups : pandas.Series or numpy.ndarray, optional
        Grouping variable for GroupKFold (e.g., cities)
    
    Returns
    -------
    dict
        Dictionary containing cross-validation results with mean and std metrics:
        - rmse_mean, rmse_std: Root Mean Square Error statistics
        - mae_mean, mae_std: Mean Absolute Error statistics  
        - r2_mean, r2_std: R-squared statistics
    """

Pipeline Integration

After completing all individual component implementations, you need to integrate them into the main pipeline execution script. The scripts/run_pipeline.py file contains a Step 4 section that requires completion to properly integrate your Evaluator implementation with the overall workflow.

Step 4: Cross-Validation Integration

Purpose: Integrate the cross-validation functionality into the main pipeline execution to evaluate model performance using geographic fold validation.

Notebook Connection: This step integrates all your previous implementations into a complete pipeline that mirrors the evaluation workflow from your original notebook.

Testing Strategy

The project includes 45 comprehensive tests that validate technical correctness across all components:

• Data Processing Tests (9 tests): Validate data loading, preprocessing, and missing value handling
• Feature Engineering Tests (11 tests): Check feature extraction, encoding, and selection methods
• Model Training Tests (14 tests): Verify model creation, training, and persistence
• Evaluation Tests (11 tests): Confirm metrics calculation and cross-validation logic

Final Validation

After completing all implementations, run the complete test suite to validate your work:

1
2
3
4
5
6
7
8

# Run all tests to ensure complete implementation
uv run python scripts/run_tests.py

# Run with verbose output for detailed feedback
uv run python scripts/run_tests.py --verbose

# Test the complete pipeline functionality
uv run python scripts/run_pipeline.py --model linear --n-features 15

What Tests Validate vs. What They Don’t

✅ Tests Validate:

• Technical Implementation: Methods execute without errors and return expected data structures
• Data Integrity: Original data is preserved through transformations
• API Contracts: Functions accept correct inputs and return expected outputs
• Error Handling: Proper exceptions for invalid inputs
• Integration: Components work together seamlessly

⚠️ Tests Don’t Validate:

• Model Performance: Tests don’t check if models achieve good RMSE/R² scores
• Data Quality: Tests use synthetic data, not real-world validation
• Statistical Significance: Tests verify functionality, not predictive accuracy
• Hyperparameter Optimization: Tests use default parameters for speed

Conclusion

This practical session has guided you through the essential process of transforming exploratory notebook code into production-ready machine learning pipelines. By completing this workshop, you have gained hands-on experience with:

• Modular Architecture: Converting monolithic notebook cells into reusable, testable components
• Software Engineering Practices: Applying separation of concerns, proper error handling, and comprehensive testing
• Production Readiness: Creating structured workflows suitable for deployment and team collaboration
• Professional Standards: Following industry best practices used by Machine Learning Engineers

The structured approach you’ve learned represents a fundamental skill for modern data scientists and ML engineers. This foundation will serve you well in professional environments where code quality, maintainability, and collaboration are essential.

Next Steps

In the next practical session, you will build upon this structured foundation to explore advanced model evaluation and comparison techniques. You’ll learn to:

• Model Selection: Systematically evaluate multiple algorithms and configurations
• Hyperparameter Optimization: Fine-tune model parameters for optimal performance
• Advanced Evaluation: Apply sophisticated validation strategies and performance metrics
• Model Comparison: Compare different approaches using statistical significance testing

This progression from structured code to advanced modeling techniques mirrors the real-world ML development process, preparing you for the challenges of professional machine learning projects.