Pipelines & Transformers

1. Why this matters

Without a pipeline, ML projects accumulate cruft: 12 lines of preprocessing in a notebook, then a model fit, then "wait, I need to do the same to the test set" — and inevitably you make a mistake. Pipelines:

• Prevent leakage — preprocessing fits inside CV folds, not before.
• Make deployment trivial — one .pkl file holds the whole flow.
• Enable hyperparameter search over preprocessing AND model parameters.
• Make code reviewable — the whole pipeline is in one place.

2. Mental model

A Pipeline is a list of (name, estimator) steps. All but the last are transformers (.fit_transform); the last is the final estimator (.fit, .predict).

flowchart LR
    X[X] --> S1[step 1: scaler<br/>fit_transform]
    S1 --> S2[step 2: encoder<br/>fit_transform]
    S2 --> S3[step 3: PCA<br/>fit_transform]
    S3 --> M[step 4: model<br/>fit]

ColumnTransformer slots in as a single step that applies different sub-transformers to different columns:

flowchart TB
    DF[DataFrame] --> CT[ColumnTransformer]
    CT --> A[num pipeline: impute, scale]
    CT --> B[cat pipeline: impute, ohe]
    CT --> C[passthrough]
    A --> JOIN[concat]
    B --> JOIN
    C --> JOIN
    JOIN --> M[Model]

3. Core concepts

• Pipeline([(name, est), ...]) — sequential steps. Last step is the final model; everything before must implement fit_transform.
• make_pipeline(t1, t2, model) — convenience that auto-names steps.
• ColumnTransformer([(name, trans, cols), ...]) — parallel sub-pipelines, one per column subset. remainder="passthrough" or "drop".
• FunctionTransformer(fn) — wraps any pure function into a transformer. Use sparingly; prefer first-class transformers when possible.
• PowerTransformer — fix skewed numeric features (Yeo-Johnson handles negatives; Box-Cox needs positives).
• Binarizer(threshold=t) — x > t → 1/0. Useful for hand-crafted binary indicators.
• KBinsDiscretizer(n_bins=k) — bucket continuous → categorical.
• set_config(transform_output="pandas") — keep DataFrame structure (column names) through the pipeline. New in sklearn 1.2+.

4. Code — minimal working example

from sklearn.pipeline import Pipeline
from sklearn.preprocessing import StandardScaler
from sklearn.linear_model import LogisticRegression

pipe = Pipeline([
    ("scale", StandardScaler()),
    ("clf",   LogisticRegression(max_iter=500)),
])

pipe.fit(X_train, y_train)
pipe.score(X_test, y_test)

# Access steps by name
pipe.named_steps["clf"].coef_

# Or via slicing
pipe[:-1].fit_transform(X_train)   # all preprocessing, no model

5. Code — real-world ColumnTransformer pipeline

The pattern you'll use in 90% of tabular ML projects:

import pandas as pd
from sklearn.compose import ColumnTransformer
from sklearn.pipeline import Pipeline
from sklearn.preprocessing import (
    StandardScaler, OneHotEncoder, OrdinalEncoder, PowerTransformer,
)
from sklearn.impute import SimpleImputer
from sklearn.ensemble import RandomForestClassifier

num_normal  = ["age", "tenure_months"]
num_skewed  = ["income", "monthly_charges"]
ord_cols    = ["education"]                          # has order
cat_cols    = ["plan_type", "country"]               # no order

num_normal_pipe = Pipeline([
    ("impute", SimpleImputer(strategy="median")),
    ("scale",  StandardScaler()),
])
num_skewed_pipe = Pipeline([
    ("impute", SimpleImputer(strategy="median")),
    ("power",  PowerTransformer(method="yeo-johnson")),
])
ord_pipe = Pipeline([
    ("impute", SimpleImputer(strategy="most_frequent")),
    ("ord",    OrdinalEncoder(
        categories=[["HS","BA","MA","PhD"]],
        handle_unknown="use_encoded_value", unknown_value=-1,
    )),
])
cat_pipe = Pipeline([
    ("impute", SimpleImputer(strategy="most_frequent")),
    ("ohe",    OneHotEncoder(handle_unknown="ignore",
                              min_frequency=10)),
])

preprocess = ColumnTransformer([
    ("num_n",  num_normal_pipe,  num_normal),
    ("num_s",  num_skewed_pipe,  num_skewed),
    ("ord",    ord_pipe,          ord_cols),
    ("cat",    cat_pipe,          cat_cols),
], remainder="drop", verbose_feature_names_out=True)

pipe = Pipeline([
    ("prep", preprocess),
    ("clf",  RandomForestClassifier(n_estimators=200, random_state=42)),
])
pipe.fit(X_train, y_train)
print("Test accuracy:", pipe.score(X_test, y_test))

# Save the whole thing
import joblib
joblib.dump(pipe, "model.joblib")

# Load and predict in production
pipe2 = joblib.load("model.joblib")
pipe2.predict(new_X)        # all preprocessing happens automatically

6. Code — FunctionTransformer for custom logic

When you need a transform that isn't first-class — log, custom math, datetime parsing:

import numpy as np
from sklearn.preprocessing import FunctionTransformer

log_transformer = FunctionTransformer(
    func=np.log1p,                   # log(1 + x)
    inverse_func=np.expm1,
    validate=True,
    feature_names_out="one-to-one",  # keep column names
)

# Use inside a pipeline like any other transformer
pipe = Pipeline([
    ("log",   log_transformer),
    ("scale", StandardScaler()),
    ("model", LinearRegression()),
])

Custom datetime extraction:

def add_datetime_features(df):
    df = df.copy()
    df["hour"]    = df["timestamp"].dt.hour
    df["weekday"] = df["timestamp"].dt.weekday
    df["is_weekend"] = (df["weekday"] >= 5).astype(int)
    return df.drop(columns=["timestamp"])

dt_step = FunctionTransformer(add_datetime_features, validate=False)

7. Code — Power transforms and binning

PowerTransformer — fix skewed features:

from sklearn.preprocessing import PowerTransformer
# Yeo-Johnson: works on positive AND negative values
pt = PowerTransformer(method="yeo-johnson", standardize=True)
X_transformed = pt.fit_transform(X[["income"]])
# now "income" looks closer to a Gaussian — better for linear models

KBinsDiscretizer — turn continuous into categorical (useful for tree-based decisions, equal-frequency bucketing):

from sklearn.preprocessing import KBinsDiscretizer

kbd = KBinsDiscretizer(
    n_bins=5,
    encode="ordinal",          # or "onehot", "onehot-dense"
    strategy="quantile",       # or "uniform", "kmeans"
)
X_binned = kbd.fit_transform(X[["age"]])

Binarizer — simple threshold to 0/1:

from sklearn.preprocessing import Binarizer
Binarizer(threshold=0.0).fit_transform(X)   # negatives → 0, positives → 1

8. Hyperparameter tuning over a pipeline

GridSearchCV can tune any nested parameter via __:

from sklearn.model_selection import GridSearchCV

param_grid = {
    "prep__num_n__scale__with_mean": [True, False],
    "clf__n_estimators": [100, 200, 400],
    "clf__max_depth":   [None, 10, 20],
}
gs = GridSearchCV(pipe, param_grid, cv=5, n_jobs=-1, scoring="f1")
gs.fit(X_train, y_train)
print(gs.best_params_, gs.best_score_)

pipename__stepname__param is how you reach into nested transformers.

9. Common pitfalls

• ❗ Fitting transformers OUTSIDE the pipeline before splitting. Defeats the whole anti-leakage benefit. Always put preprocessing INSIDE the pipeline.
• ❗ Mixing pd.get_dummies (manual) with a pipeline. Production breaks when test data has a new category. Use OneHotEncoder in the pipeline.
• ❗ Forgetting handle_unknown="ignore" in deployed pipelines. New category at inference → crash.
• ❗ FunctionTransformer with side effects. Functions should be pure and deterministic — no global state, no random number generation without a seed.
• ❗ remainder="passthrough" carrying leftover unwanted columns into the model. Either pass them through intentionally or set remainder="drop".
• ❗ Using Pipeline but still calling .fit_transform() on each step manually. Defeats the abstraction; one bug and you've leaked.
• ❗ Pickle compatibility across sklearn versions. A .joblib made with sklearn 1.3 may not load in 1.5. Pin versions; retrain on upgrade.

10. When to use vs not use

Use	When
Pipeline	Always for any project beyond a toy notebook.
ColumnTransformer	Different transforms per column subset (numeric vs categorical).
make_pipeline	Quick prototyping — auto-names steps.
FunctionTransformer	Custom logic that doesn't fit any built-in transformer.
PowerTransformer	Skewed numeric features (right-tail income, prices).
KBinsDiscretizer	Make non-linear behavior tractable for linear models; convert age → age-bucket.
Skip all of this	One-shot script that fits and predicts in 10 lines — but you'll regret it.

11. Cheatsheet

from sklearn.pipeline    import Pipeline, make_pipeline, FeatureUnion
from sklearn.compose     import ColumnTransformer, make_column_transformer, make_column_selector

# Sequential composition
pipe = Pipeline([("scale", StandardScaler()), ("model", LogisticRegression())])
pipe = make_pipeline(StandardScaler(), LogisticRegression())     # auto-names

# Per-column subpipelines
ct = ColumnTransformer([
    ("num", num_pipe, ["age", "income"]),
    ("cat", cat_pipe, ["country"]),
], remainder="drop", verbose_feature_names_out=True)

# Selector by dtype (avoid hard-coded column names)
from sklearn.compose import make_column_selector
ct = make_column_transformer(
    (StandardScaler(), make_column_selector(dtype_include="number")),
    (OneHotEncoder(handle_unknown="ignore"),
     make_column_selector(dtype_include="object")),
)

# Inspect
pipe.named_steps                          # dict of steps
pipe.named_steps["clf"].coef_
pipe[:-1].fit_transform(X)                # transform-only (no final model)
pipe.get_feature_names_out()              # column names after all transforms

# Output as DataFrame (sklearn 1.2+)
from sklearn import set_config
set_config(transform_output="pandas")

# Persistence
import joblib
joblib.dump(pipe, "model.joblib")
pipe = joblib.load("model.joblib")

# Custom function as transformer
from sklearn.preprocessing import FunctionTransformer
log = FunctionTransformer(np.log1p, inverse_func=np.expm1,
                          feature_names_out="one-to-one")

12. Q&A — recall test

• Q: Why use a Pipeline instead of preprocessing manually? A: It prevents data leakage during CV (preprocessing refits per fold), enables one-shot save/load, and lets you tune preprocessing hyperparameters jointly with the model.

• Q: Difference between Pipeline and ColumnTransformer? A: Pipeline is sequential — each step takes the previous output. ColumnTransformer is parallel — different transformers for different columns, results concatenated.

• Q: How do you access a nested hyperparameter for tuning? A: "step_name__param". Deeper: "outer__middle__inner_param". Example: "clf__n_estimators", "prep__num__scale__with_mean".

• Q: When is FunctionTransformer the right answer? A: When no built-in transformer does what you need (e.g., custom datetime extraction, business rules) AND the function is pure. For one-liners over a column, often Pipeline + FunctionTransformer is cleaner than a custom transformer class.

• Q: How do you keep DataFrame column names through a pipeline? A: from sklearn import set_config; set_config(transform_output="pandas"). Then transformers return DataFrames instead of numpy arrays.

• Q: How to handle a new category at inference time? A: OneHotEncoder(handle_unknown="ignore") + the pipeline is saved as one artifact. New categories at inference become all-zeros and don't crash.

Practice

What does this print?

Expected: True

from sklearn.pipeline import Pipeline
from sklearn.preprocessing import StandardScaler
from sklearn.linear_model import LogisticRegression
pipe = Pipeline([("scale", StandardScaler()), ("clf", LogisticRegression())])
print("scale" in dict(pipe.steps) and "clf" in dict(pipe.steps))

Apply different preprocessing to numeric vs categorical columns (use ColumnTransformer)

Expected: True

from sklearn.pipeline import Pipeline
from sklearn.preprocessing import StandardScaler, OneHotEncoder
pipe = Pipeline([
    ("scale", StandardScaler()),         # bug: scales EVERYTHING — including categorical columns
    ("ohe", OneHotEncoder()),
])
print(hasattr(pipe, "steps"))

Quiz — Quick check

What you remember

Q1. What's the main reason to use Pipeline instead of separate steps?

• Faster training
• Prevents data leakage during cross-validation by fitting preprocessing only on training folds
• Smaller models
• Required by sklearn

Why: Without Pipeline, a manually-fit scaler from the full data leaks into every CV fold. Pipeline fits preprocessing inside each fold automatically — leak-free CV scores.

Q2. What does ColumnTransformer do?

• Combines multiple models
• Applies different transformers to different sets of columns
• Drops columns
• Normalizes a column

Why: Real data has mixed types. ColumnTransformer lets you scale numerics, one-hot categoricals, impute differently per column type — all in one composable step that fits into a Pipeline.

Q3. Why save the entire pipeline (preprocessing + model) instead of just the model?

• To save disk space
• Inference needs the exact same preprocessing — saving the pipeline guarantees the production code applies it consistently
• Required for cloud deployment
• Improves accuracy

Why: A model trained on scaled data needs scaled inputs at inference. Saving just the model means rebuilding the scaler in the serving code — error-prone. Save the pipeline; load it; call .predict() — done.

Common doubts

When should I use make_pipeline vs Pipeline?

make_pipeline(StandardScaler(), LogisticRegression()) auto-names steps (standardscaler, logisticregression). Pipeline([("scale", StandardScaler()), ("clf", ...)]) gives explicit names — useful when referencing steps via pipe.named_steps["scale"] or in grid search params.

Can I add custom steps to a Pipeline?

Yes. Subclass BaseEstimator, TransformerMixin and implement fit, transform. Or use FunctionTransformer(my_function) for stateless logic. Custom steps integrate seamlessly with Pipeline and ColumnTransformer.

How do I tune hyperparameters across the pipeline?

Use double-underscore syntax in GridSearchCV's param_grid: {"clf__C": [0.1, 1, 10], "scale__with_mean": [True, False]}. The prefix is the step name, then __, then the parameter.