Scikit-Learn Pipelines

Real-World Scenario

A data scientist builds a customer churn model. The dataset has numeric features that need scaling, categorical features that need one-hot encoding, and missing values that need imputation. Without a Pipeline, they manually apply each step before train/test split — accidentally fitting the scaler on test data and inflating the evaluation metric by 4%. The model ships, performs worse than expected, and the team spends a week debugging. Pipelines make this class of bug impossible.

Basic Pipeline

from sklearn.pipeline import Pipeline
from sklearn.preprocessing import StandardScaler
from sklearn.linear_model import LogisticRegression
from sklearn.datasets import load_breast_cancer
from sklearn.model_selection import train_test_split

X, y = load_breast_cancer(return_X_y=True)
X_train, X_test, y_train, y_test = train_test_split(X, y, test_size=0.2, random_state=42)

# Pipeline: named steps executed in order
# Each step except the last must implement fit_transform
# The last step must implement fit and predict
pipeline = Pipeline([
    ("scaler",     StandardScaler()),         # step 1: scale features
    ("classifier", LogisticRegression(max_iter=1000)),  # step 2: train model
])

# fit() calls scaler.fit_transform(X_train) then classifier.fit(X_train_scaled, y_train)
pipeline.fit(X_train, y_train)

# predict() calls scaler.transform(X_test) then classifier.predict(X_test_scaled)
accuracy = pipeline.score(X_test, y_test)
print(f"Pipeline accuracy: {accuracy:.4f}")

# Access individual steps
scaler = pipeline.named_steps["scaler"]
print(f"Feature mean (first 3): {scaler.mean_[:3].round(2)}")

ColumnTransformer — Different Preprocessing per Column

Real-world data has mixed column types. ColumnTransformer applies different transformations to different subsets of columns.

import pandas as pd
import numpy as np
from sklearn.pipeline import Pipeline
from sklearn.compose import ColumnTransformer
from sklearn.preprocessing import StandardScaler, OneHotEncoder
from sklearn.impute import SimpleImputer
from sklearn.ensemble import GradientBoostingClassifier
from sklearn.model_selection import train_test_split
from sklearn.metrics import classification_report

rng = np.random.default_rng(42)
n = 1000

df = pd.DataFrame({
    "age":          rng.integers(18, 70, n).astype(float),
    "salary":       rng.uniform(20000, 200000, n),
    "tenure_years": rng.integers(0, 20, n).astype(float),
    "department":   rng.choice(["Engineering", "Sales", "Marketing", "HR"], n),
    "employment":   rng.choice(["full-time", "part-time", "contract"], n),
    "churned":      rng.integers(0, 2, n),
})

# Inject missing values
df.loc[rng.choice(n, 50), "age"] = np.nan
df.loc[rng.choice(n, 30), "salary"] = np.nan

X = df.drop("churned", axis=1)
y = df["churned"]

X_train, X_test, y_train, y_test = train_test_split(X, y, test_size=0.2, random_state=42)

# Define column groups
numeric_cols     = ["age", "salary", "tenure_years"]
categorical_cols = ["department", "employment"]

# Preprocessing pipeline per column type
numeric_transformer = Pipeline([
    ("imputer", SimpleImputer(strategy="median")),   # fill missing with median
    ("scaler",  StandardScaler()),
])

categorical_transformer = Pipeline([
    ("imputer", SimpleImputer(strategy="most_frequent")),  # fill missing with mode
    ("encoder", OneHotEncoder(handle_unknown="ignore", sparse_output=False)),
])

# Combine into a ColumnTransformer
preprocessor = ColumnTransformer([
    ("num", numeric_transformer,    numeric_cols),
    ("cat", categorical_transformer, categorical_cols),
])

# Full pipeline: preprocess + model
full_pipeline = Pipeline([
    ("preprocessor", preprocessor),
    ("model",        GradientBoostingClassifier(n_estimators=200, random_state=42)),
])

full_pipeline.fit(X_train, y_train)
print(classification_report(y_test, full_pipeline.predict(X_test)))

Grid Search with Pipeline

from sklearn.pipeline import Pipeline
from sklearn.preprocessing import StandardScaler
from sklearn.svm import SVC
from sklearn.model_selection import GridSearchCV, train_test_split
from sklearn.datasets import load_iris

X, y = load_iris(return_X_y=True)
X_train, X_test, y_train, y_test = train_test_split(X, y, test_size=0.2, random_state=42)

pipe = Pipeline([
    ("scaler", StandardScaler()),
    ("svm",    SVC()),
])

# Parameter grid — use step_name__param_name syntax
param_grid = {
    "scaler__with_mean": [True, False],    # whether to center data
    "svm__C":            [0.1, 1, 10],    # regularization strength
    "svm__kernel":       ["linear", "rbf"],
    "svm__gamma":        ["scale", "auto"],
}

search = GridSearchCV(
    pipe, param_grid,
    cv=5,                   # 5-fold cross-validation
    scoring="accuracy",
    n_jobs=-1,              # use all CPU cores
    verbose=1,
)
search.fit(X_train, y_train)

print(f"Best params:  {search.best_params_}")
print(f"Best CV acc:  {search.best_score_:.4f}")
print(f"Test acc:     {search.score(X_test, y_test):.4f}")

Custom Transformers

import numpy as np
import pandas as pd
from sklearn.base import BaseEstimator, TransformerMixin
from sklearn.pipeline import Pipeline
from sklearn.preprocessing import StandardScaler

class LogTransformer(BaseEstimator, TransformerMixin):
    """Apply log1p to skewed numeric columns."""

    def __init__(self, columns=None):
        self.columns = columns

    def fit(self, X, y=None):
        # Nothing to learn — return self
        return self

    def transform(self, X):
        X = X.copy()
        cols = self.columns or X.columns.tolist() if hasattr(X, "columns") else range(X.shape[1])
        for col in cols:
            X[col] = np.log1p(X[col].clip(lower=0))
        return X


# Use in a pipeline just like any built-in transformer
rng = np.random.default_rng(42)
df = pd.DataFrame({
    "salary":  rng.exponential(50000, 500),   # right-skewed
    "age":     rng.integers(18, 65, 500).astype(float),
    "churned": rng.integers(0, 2, 500),
})

X = df.drop("churned", axis=1)
y = df["churned"]

pipe = Pipeline([
    ("log_transform", LogTransformer(columns=["salary"])),
    ("scaler",        StandardScaler()),
])

X_transformed = pipe.fit_transform(X)
print(X_transformed.shape)  # (500, 2)