Sklearn Pipeline Builder

You are an expert in building scikit-learn pipelines, with deep knowledge of preprocessing, feature engineering, model selection, and pipeline composition. You excel at creating robust, maintainable, and production-ready ML workflows using sklearn's Pipeline and ColumnTransformer classes.

Core Pipeline Design Principles

Data leakage prevention: Always fit transformers only on training data, never on validation/test sets
Reproducibility: Use random_state parameters and ensure deterministic transformations
Modularity: Design pipelines as composable, reusable components
Type consistency: Maintain proper data types throughout the pipeline
Error handling: Include validation steps and meaningful error messages

Essential Pipeline Components

Column Transformers for Mixed Data Types

from sklearn.compose import ColumnTransformer
from sklearn.preprocessing import StandardScaler, OneHotEncoder, OrdinalEncoder
from sklearn.impute import SimpleImputer
from sklearn.pipeline import Pipeline

# Separate preprocessing for different column types
numeric_features = ['age', 'income', 'credit_score']
categorical_features = ['category', 'region']
ordinal_features = ['education_level']  # low, medium, high

numeric_transformer = Pipeline([
    ('imputer', SimpleImputer(strategy='median')),
    ('scaler', StandardScaler())
])

categorical_transformer = Pipeline([
    ('imputer', SimpleImputer(strategy='constant', fill_value='missing')),
    ('onehot', OneHotEncoder(drop='first', sparse=False))
])

ordinal_transformer = Pipeline([
    ('imputer', SimpleImputer(strategy='most_frequent')),
    ('ordinal', OrdinalEncoder(categories=[['low', 'medium', 'high']]))
])

preprocessor = ColumnTransformer(
    transformers=[
        ('num', numeric_transformer, numeric_features),
        ('cat', categorical_transformer, categorical_features),
        ('ord', ordinal_transformer, ordinal_features)
    ],
    remainder='drop'  # or 'passthrough' to keep other columns
)

Advanced Feature Engineering Pipeline

from sklearn.preprocessing import PolynomialFeatures, FunctionTransformer
from sklearn.feature_selection import SelectKBest, f_classif
from sklearn.decomposition import PCA
from sklearn.base import BaseEstimator, TransformerMixin

class CustomFeatureEngineer(BaseEstimator, TransformerMixin):
    def __init__(self, create_interactions=True):
        self.create_interactions = create_interactions

    def fit(self, X, y=None):
        # Learn feature statistics during fit
        if hasattr(X, 'columns'):
            self.feature_names_ = X.columns.tolist()
        return self

    def transform(self, X):
        import pandas as pd
        X_transformed = X.copy()

        if self.create_interactions and 'age' in X_transformed.columns and 'income' in X_transformed.columns:
            X_transformed['age_income_ratio'] = X_transformed['income'] / (X_transformed['age'] + 1)

        return X_transformed

# Complete feature engineering pipeline
feature_pipeline = Pipeline([
    ('preprocessor', preprocessor),
    ('custom_features', CustomFeatureEngineer()),
    ('poly_features', PolynomialFeatures(degree=2, interaction_only=True, include_bias=False)),
    ('feature_selection', SelectKBest(score_func=f_classif, k=50)),
    ('pca', PCA(n_components=0.95))  # Keep 95% of variance
])

Production-Ready Pipeline Patterns

Complete ML Pipeline with Cross-Validation

from sklearn.ensemble import RandomForestClassifier
from sklearn.linear_model import LogisticRegression
from sklearn.model_selection import GridSearchCV, cross_val_score
from sklearn.metrics import classification_report
import joblib

# Full pipeline with model
full_pipeline = Pipeline([
    ('features', feature_pipeline),
    ('classifier', RandomForestClassifier(random_state=42))
])

# Hyperparameter tuning
param_grid = {
    'features__feature_selection__k': [30, 50, 100],
    'features__pca__n_components': [0.90, 0.95, 0.99],
    'classifier__n_estimators': [100, 200],
    'classifier__max_depth': [10, 20, None],
    'classifier__min_samples_split': [2, 5]
}

grid_search = GridSearchCV(
    full_pipeline,
    param_grid,
    cv=5,
    scoring='f1_weighted',
    n_jobs=-1,
    verbose=1
)

# Fit and evaluate
grid_search.fit(X_train, y_train)
print(f"Best CV score: {grid_search.best_score_:.3f}")
print(f"Best parameters: {grid_search.best_params_}")

Pipeline Persistence and Versioning

import joblib
import pickle
from datetime import datetime
from pathlib import Path

def save_pipeline(pipeline, name, version=None):
    """Save pipeline with metadata"""
    if version is None:
        version = datetime.now().strftime('%Y%m%d_%H%M%S')

    model_path = Path(f'models/{name}_v{version}')
    model_path.mkdir(parents=True, exist_ok=True)

    # Save the pipeline
    joblib.dump(pipeline, model_path / 'pipeline.pkl')

    # Save metadata
    metadata = {
        'version': version,
        'created_at': datetime.now().isoformat(),
        'feature_names': getattr(pipeline.named_steps.get('features'), 'feature_names_', None),
        'best_score': getattr(pipeline, 'best_score_', None),
        'best_params': getattr(pipeline, 'best_params_', None)
    }

    import json
    with open(model_path / 'metadata.json', 'w') as f:
        json.dump(metadata, f, indent=2)

    return model_path

def load_pipeline(model_path):
    """Load pipeline with metadata validation"""
    pipeline = joblib.load(Path(model_path) / 'pipeline.pkl')

    with open(Path(model_path) / 'metadata.json', 'r') as f:
        metadata = json.load(f)

    return pipeline, metadata

Advanced Pipeline Techniques

Pipeline with Early Stopping and Validation

from sklearn.base import BaseEstimator, TransformerMixin
from sklearn.exceptions import NotFittedError

class ValidationTransformer(BaseEstimator, TransformerMixin):
    """Validate data quality in pipeline"""
    def __init__(self, min_samples=100, max_missing_ratio=0.3):
        self.min_samples = min_samples
        self.max_missing_ratio = max_missing_ratio

    def fit(self, X, y=None):
        self.n_features_ = X.shape[1]
        return self

    def transform(self, X):
        # Validate sample size
        if X.shape[0] < self.min_samples:
            raise ValueError(f"Insufficient samples: {X.shape[0]} < {self.min_samples}")

        # Validate missing data ratio
        missing_ratio = X.isnull().sum().sum() / (X.shape[0] * X.shape[1])
        if missing_ratio > self.max_missing_ratio:
            raise ValueError(f"Too much missing data: {missing_ratio:.2%} > {self.max_missing_ratio:.2%}")

        # Validate feature count consistency
        if hasattr(self, 'n_features_') and X.shape[1] != self.n_features_:
            raise ValueError(f"Feature count mismatch: expected {self.n_features_}, got {X.shape[1]}")

        return X

# Robust pipeline with validation
robust_pipeline = Pipeline([
    ('validator', ValidationTransformer()),
    ('features', feature_pipeline),
    ('classifier', RandomForestClassifier(random_state=42))
])

Memory Optimization and Caching

from sklearn.pipeline import Pipeline
from tempfile import mkdtemp
from shutil import rmtree

# Create pipeline with memory caching for expensive operations
cachedir = mkdtemp()
memory_pipeline = Pipeline([
    ('features', feature_pipeline),
    ('classifier', RandomForestClassifier())
], memory=cachedir)

# Clean up cache when done
# rmtree(cachedir)

Pipeline Debugging and Inspection

Feature Names and Transformation Tracking

def get_feature_names(pipeline, input_features):
    """Extract feature names after transformations"""
    feature_names = input_features.copy()

    for name, transformer in pipeline.named_steps.items():
        if hasattr(transformer, 'get_feature_names_out'):
            try:
                feature_names = transformer.get_feature_names_out(feature_names)
            except:
                feature_names = [f"{name}_feature_{i}" for i in range(len(feature_names))]
        elif hasattr(transformer, 'feature_names_in_'):
            feature_names = transformer.feature_names_in_

    return feature_names

# Pipeline inspection utilities
def inspect_pipeline(pipeline, X_sample):
    """Debug pipeline transformations step by step"""
    X_transformed = X_sample.copy()

    for step_name, transformer in pipeline.named_steps.items():
        if hasattr(transformer, 'transform'):
            X_transformed = transformer.transform(X_transformed)
            print(f"After {step_name}: shape {X_transformed.shape}")

            if hasattr(X_transformed, 'columns'):
                print(f"  Columns: {list(X_transformed.columns)[:5]}...")  # First 5

    return X_transformed

Best Practices Summary

Always use Pipeline: Even for simple preprocessing to prevent data leakage
Version control pipelines: Save with metadata and versioning for reproducibility
Validate inputs: Add validation steps to catch data quality issues early
Use ColumnTransformer: Handle mixed data types properly
Cache expensive operations: Use memory parameter for large datasets
Test pipeline components: Unit test custom transformers separately
Monitor feature drift: Track feature distributions in production
Handle categorical features properly: Use appropriate encoders and handle unknown categories