Scikit-Learn Pipelines

A pipeline chains preprocessing steps and a model into one estimator, so fit and predict run every step in the right order on the right data. Every example here ran with scikit-learn 1.9.1.

Why Pipelines Exist: Leakage

These labels are random coin flips, so no honest model can beat about 50%. The first evaluation selects the “best” features on all rows before cross-validating.

import numpy as np
from sklearn.feature_selection import SelectKBest, f_classif
from sklearn.linear_model import LogisticRegression
from sklearn.model_selection import cross_val_score
from sklearn.pipeline import make_pipeline

rng = np.random.default_rng(0)
X = rng.standard_normal((100, 5000))      # 5000 pure-noise features
y = rng.integers(0, 2, 100)               # random labels

X_sel = SelectKBest(f_classif, k=20).fit_transform(X, y)    # sees every row
leaky = cross_val_score(LogisticRegression(), X_sel, y, cv=5).mean()

pipe = make_pipeline(SelectKBest(f_classif, k=20), LogisticRegression())
honest = cross_val_score(pipe, X, y, cv=5).mean()

print(f"select, then cross-validate:  {leaky:.3f}")
print(f"select inside each fold:      {honest:.3f}")

$ python leakage.py
select, then cross-validate:  0.850
select inside each fold:      0.480

With 5000 noise features, some correlate with the labels by chance, including in the rows later used for testing. Selecting them first bakes the test labels into the features, so the model scores far above chance on data with no signal at all. Inside the pipeline, selection only sees each fold’s training rows, and the score falls back to a coin flip.

A Basic Pipeline

Pipeline takes named steps. Every step but the last must transform; the last one predicts.

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

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)

pipe = Pipeline([
    ("scaler", StandardScaler()),
    ("clf", LogisticRegression(max_iter=1000)),
])
pipe.fit(X_train, y_train)                 # scaler.fit_transform, then clf.fit
print(f"test accuracy: {pipe.score(X_test, y_test):.4f}")
print("scaler means (first 3):", pipe.named_steps["scaler"].mean_[:3].round(2))

$ python basic_pipeline.py
test accuracy: 0.9737
scaler means (first 3): [14.12 19.19 91.88]

fit learned the scaler’s means from the 455 training rows only; score reused them to transform the test rows before predicting. named_steps gives access to each fitted step.

ColumnTransformer for Mixed Data

Real tables mix numbers, categories and gaps. ColumnTransformer routes each group of columns to its own preprocessing.

import numpy as np
import pandas as pd

rng = np.random.default_rng(42)
n = 1000
df = pd.DataFrame({
    "tenure": rng.integers(0, 20, n).astype(float),
    "salary": rng.uniform(20_000, 200_000, n),
    "contract": rng.choice(["monthly", "yearly"], n),
    "dept": rng.choice(["Sales", "Eng", "HR"], n),
})
# churn is likelier for short-tenure monthly contracts
p = 0.15 + 0.6 * ((df["tenure"] < 4) & (df["contract"] == "monthly"))
df["churned"] = (rng.random(n) < p).astype(int)
df.loc[rng.choice(n, 60, replace=False), "tenure"] = np.nan    # gaps
df.to_csv("churn.csv", index=False)
print(df.head(3))

$ python mixed_data.py
   tenure         salary contract   dept  churned
0     NaN  152607.023927   yearly  Sales        1
1    15.0  179552.519610   yearly     HR        0
2    13.0  185790.295513   yearly  Sales        0

That builds a small churn table with a planted pattern and some missing tenure values. Now a single pipeline takes the raw DataFrame to predictions.

import pandas as pd
from sklearn.compose import ColumnTransformer
from sklearn.ensemble import GradientBoostingClassifier
from sklearn.impute import SimpleImputer
from sklearn.model_selection import train_test_split
from sklearn.pipeline import make_pipeline
from sklearn.preprocessing import OneHotEncoder, StandardScaler

df = pd.read_csv("churn.csv")
X, y = df.drop(columns="churned"), df["churned"]
X_train, X_test, y_train, y_test = train_test_split(X, y, test_size=0.25, random_state=0)

pre = ColumnTransformer([
    ("num", make_pipeline(SimpleImputer(strategy="median"), StandardScaler()),
     ["tenure", "salary"]),
    ("cat", OneHotEncoder(handle_unknown="ignore"), ["contract", "dept"]),
])
model = make_pipeline(pre, GradientBoostingClassifier(random_state=0))
model.fit(X_train, y_train)
print(f"test accuracy: {model.score(X_test, y_test):.3f}")
print(pre.get_feature_names_out().tolist())

$ python column_transformer.py
test accuracy: 0.796
['num__tenure', 'num__salary', 'cat__contract_monthly', 'cat__contract_yearly', 'cat__dept_Eng', 'cat__dept_HR', 'cat__dept_Sales']

The numeric branch imputes then scales; the categorical branch one-hot encodes; the outputs are concatenated in the order listed in get_feature_names_out. Because the whole thing is one estimator, model.predict(new_raw_dataframe) repeats every step with the values learned from the training rows.

Grid Search Over a Pipeline

Parameters inside a pipeline are addressed as step__parameter.

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

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)

pipe = Pipeline([("scaler", StandardScaler()), ("svm", SVC())])
grid = {"svm__C": [0.1, 1, 10], "svm__kernel": ["linear", "rbf"]}

search = GridSearchCV(pipe, grid, cv=5).fit(X_train, y_train)
print("best params:", search.best_params_)
print(f"best CV accuracy: {search.best_score_:.4f}")
print(f"test accuracy:    {search.score(X_test, y_test):.4f}")

$ python grid_pipeline.py
best params: {'svm__C': 0.1, 'svm__kernel': 'linear'}
best CV accuracy: 0.9736
test accuracy:    0.9825

Six combinations times five folds is 30 fits, and the scaler is refit inside every one. After the search, search refits the best pipeline on all training rows, which is what search.score evaluates on the untouched test set.

A Custom Transformer

A transformer learns something in fit and applies it in transform. This one learns percentile limits from the training data and clips outliers to them.

import numpy as np
from sklearn.base import BaseEstimator, TransformerMixin

class PercentileClipper(BaseEstimator, TransformerMixin):
    def __init__(self, lower=5, upper=95):
        self.lower = lower
        self.upper = upper

    def fit(self, X, y=None):
        self.low_ = np.percentile(X, self.lower, axis=0)     # learned state ends in _
        self.high_ = np.percentile(X, self.upper, axis=0)
        return self

    def transform(self, X):
        return np.clip(X, self.low_, self.high_)

train = np.arange(1, 101, dtype=float).reshape(-1, 1)     # 1..100
clipper = PercentileClipper().fit(train)
print("learned limits:", clipper.low_, clipper.high_)
print(clipper.transform(np.array([[-500.0], [50.0], [9999.0]])).ravel())

$ python custom_transformer.py
learned limits: [5.95] [95.05]
[ 5.95 50.   95.05]

The limits came from the training data (the 5th and 95th percentiles of 1-100), and transform applied them to new values it never saw. Because it follows the contract, the class works inside Pipeline, GridSearchCV (clipper__upper is tunable) and clone.

Practice

import numpy as np
from sklearn.feature_selection import SelectKBest, f_classif
from sklearn.linear_model import LogisticRegression
from sklearn.model_selection import cross_val_score
from sklearn.pipeline import make_pipeline

rng = np.random.default_rng(0)
X = rng.standard_normal((100, 5000))
y = rng.integers(0, 2, 100)

for k in [20, 200]:
    X_sel = SelectKBest(f_classif, k=k).fit_transform(X, y)
    leaky = cross_val_score(LogisticRegression(), X_sel, y, cv=5).mean()
    honest = cross_val_score(make_pipeline(SelectKBest(f_classif, k=k),
                                           LogisticRegression()), X, y, cv=5).mean()
    print(f"k={k:3d}  leaky {leaky:.3f}  honest {honest:.3f}")

$ python p1_leak_k.py
k= 20  leaky 0.850  honest 0.480
k=200  leaky 0.990  honest 0.400

import pandas as pd
from sklearn.compose import ColumnTransformer
from sklearn.ensemble import GradientBoostingClassifier
from sklearn.impute import SimpleImputer
from sklearn.model_selection import GridSearchCV
from sklearn.pipeline import Pipeline
from sklearn.preprocessing import OneHotEncoder

df = pd.read_csv("churn.csv")
X, y = df.drop(columns="churned"), df["churned"]

pre = ColumnTransformer([
    ("num", Pipeline([("imputer", SimpleImputer())]), ["tenure", "salary"]),
    ("cat", OneHotEncoder(handle_unknown="ignore"), ["contract", "dept"]),
])
pipe = Pipeline([("pre", pre), ("gb", GradientBoostingClassifier(random_state=0))])
grid = {"pre__num__imputer__strategy": ["mean", "median", "most_frequent"]}
search = GridSearchCV(pipe, grid, cv=5).fit(X, y)
for s, m in zip(search.cv_results_["param_pre__num__imputer__strategy"],
                search.cv_results_["mean_test_score"]):
    print(f"{s:14s} {m:.4f}")

$ python p2_nested_param.py
mean           0.8010
median         0.7910
most_frequent  0.8020

import numpy as np
from sklearn.preprocessing import FunctionTransformer

rng = np.random.default_rng(0)
salary = rng.exponential(50_000, size=(1000, 1))      # right-skewed

log = FunctionTransformer(np.log1p, inverse_func=np.expm1)
out = log.fit_transform(salary)
print(f"raw:  mean {salary.mean():>9.0f}  median {np.median(salary):>9.0f}")
print(f"log1p: mean {out.mean():>8.2f}  median {np.median(out):>8.2f}")
print("round trip:", np.allclose(log.inverse_transform(out), salary))

$ python p3_function_transformer.py
raw:  mean     51102  median     37173
log1p: mean    10.28  median    10.52
round trip: True

import numpy as np
from sklearn.base import BaseEstimator, TransformerMixin
from sklearn.datasets import load_diabetes
from sklearn.linear_model import Ridge
from sklearn.model_selection import GridSearchCV
from sklearn.pipeline import Pipeline

class PercentileClipper(BaseEstimator, TransformerMixin):
    def __init__(self, lower=5, upper=95):
        self.lower, self.upper = lower, upper
    def fit(self, X, y=None):
        self.low_ = np.percentile(X, self.lower, axis=0)
        self.high_ = np.percentile(X, self.upper, axis=0)
        return self
    def transform(self, X):
        return np.clip(X, self.low_, self.high_)

X, y = load_diabetes(return_X_y=True)
pipe = Pipeline([("clip", PercentileClipper()), ("ridge", Ridge())])
search = GridSearchCV(pipe, {"clip__upper": [90, 95, 99, 100]}, cv=5).fit(X, y)
print("best:", search.best_params_, f"R2 {search.best_score_:.4f}")

$ python p4_tune_custom.py
best: {'clip__upper': 100} R2 0.4094