Transformation Pipelines & Custom Transformers

What you’ll learn

what goes wrong when preprocessing is a sequence of loose steps
PipelinePipeline and ColumnTransformerColumnTransformer, and the one rule that makes them leak-proof
how to write a transformer scikit-learn will accept, in about fifteen lines
leakage demonstrated: 0.867 accuracy from a dataset with no signal in it at all
the leaky-scaler arithmetic, computed by hand
how fitfit and transformtransform propagate through a nested pipeline

Intuition

By now the preprocessing is: impute the medians, add three ratio columns, one-hot the categorical column, scale the numeric ones. Four steps, each with learned parameters — nine medians, five category names, nine means and nine standard deviations.

Done by hand, every one of those has to be recreated identically for the test set, for every cross-validation fold, and for every live request. Miss one and the model receives inputs that do not match what it was trained on. Recompute one on the wrong data and you have leaked.

A PipelinePipeline makes that structural. The whole graph becomes a single estimator with one fitfit and one transformtransform, and the parameters are learned exactly once, from exactly the right rows.

figureThe whole preprocessing graph, as one objectmatplotlib

Two branches, one output. fit() learns every parameter from the training fold only; predict() replays them. Nothing else in this phase prevents leakage as reliably.

The problem with loose steps

by_hand.py

# Every one of these has to be repeated, identically, for the test set
imputer = SimpleImputer(strategy="median").fit(X_train_num)
X_train_num_imp = imputer.transform(X_train_num)
 
X_train_num_imp = add_ratio_features(X_train_num_imp)
 
scaler = StandardScaler().fit(X_train_num_imp)
X_train_num_scaled = scaler.transform(X_train_num_imp)
 
encoder = OneHotEncoder(handle_unknown="ignore").fit(X_train_cat)
X_train_cat_enc = encoder.transform(X_train_cat)
 
X_train_prepared = np.hstack([X_train_num_scaled, X_train_cat_enc.toarray()])
# ... and now all of it again for X_test, in the same order, with the same objects

by_hand.py

# Every one of these has to be repeated, identically, for the test set
imputer = SimpleImputer(strategy="median").fit(X_train_num)
X_train_num_imp = imputer.transform(X_train_num)
 
X_train_num_imp = add_ratio_features(X_train_num_imp)
 
scaler = StandardScaler().fit(X_train_num_imp)
X_train_num_scaled = scaler.transform(X_train_num_imp)
 
encoder = OneHotEncoder(handle_unknown="ignore").fit(X_train_cat)
X_train_cat_enc = encoder.transform(X_train_cat)
 
X_train_prepared = np.hstack([X_train_num_scaled, X_train_cat_enc.toarray()])
# ... and now all of it again for X_test, in the same order, with the same objects

Four fitted objects to keep in sync, one hard-coded column order, and no way to cross-validate the whole thing — because cross_val_scorecross_val_score would need to redo the preprocessing inside each fold, and it cannot reach into this code.

Pipeline

pipeline.py

from sklearn.impute import SimpleImputer
from sklearn.pipeline import Pipeline
from sklearn.preprocessing import StandardScaler
 
num_pipeline = Pipeline([
    ("imputer", SimpleImputer(strategy="median")),
    ("attribs", AttributeAdder()),
    ("scaler", StandardScaler()),
])
 
X_train_num_prepared = num_pipeline.fit_transform(X_train_num)
X_test_num_prepared = num_pipeline.transform(X_test_num)     # transform only

pipeline.py

from sklearn.impute import SimpleImputer
from sklearn.pipeline import Pipeline
from sklearn.preprocessing import StandardScaler
 
num_pipeline = Pipeline([
    ("imputer", SimpleImputer(strategy="median")),
    ("attribs", AttributeAdder()),
    ("scaler", StandardScaler()),
])
 
X_train_num_prepared = num_pipeline.fit_transform(X_train_num)
X_test_num_prepared = num_pipeline.transform(X_test_num)     # transform only

The contract: every step except the last must be a transformer (fitfit and transformtransform); the last may be a transformer or a predictor. Calling fitfit runs fit_transformfit_transform down the chain and fitfit on the final step. Calling transformtransform or predictpredict runs transformtransform down the chain.

Access the fitted pieces by name — num_pipeline.named_steps["imputer"].statistics_num_pipeline.named_steps["imputer"].statistics_ — or by position, num_pipeline[0]num_pipeline[0].

A custom transformer

Any class with fitfit and transformtransform works. Inheriting from BaseEstimatorBaseEstimator and TransformerMixinTransformerMixin adds get_paramsget_params / set_paramsset_params (so GridSearchCVGridSearchCV can tune it) and fit_transformfit_transform (so you do not write it yourself):

attribute_adder.py

import numpy as np
from sklearn.base import BaseEstimator, TransformerMixin
 
ROOMS, BEDROOMS, POPULATION, HOUSEHOLDS = 3, 4, 5, 6      # column indices
 
 
class AttributeAdder(BaseEstimator, TransformerMixin):
    """Add the three ratio features found during exploration."""
 
    def __init__(self, add_bedrooms_per_room=True):
        # No validation and no *args/**kwargs here — get_params relies on
        # every constructor argument being stored under its own name.
        self.add_bedrooms_per_room = add_bedrooms_per_room
 
    def fit(self, X, y=None):
        return self          # nothing to learn; still required by the API
 
    def transform(self, X):
        rooms_per_household = X[:, ROOMS] / X[:, HOUSEHOLDS]
        population_per_household = X[:, POPULATION] / X[:, HOUSEHOLDS]
        if self.add_bedrooms_per_room:
            bedrooms_per_room = X[:, BEDROOMS] / X[:, ROOMS]
            return np.c_[X, rooms_per_household, population_per_household,
                         bedrooms_per_room]
        return np.c_[X, rooms_per_household, population_per_household]
 
 
adder = AttributeAdder(add_bedrooms_per_room=False)
print(adder.get_params())     # {'add_bedrooms_per_room': False}

attribute_adder.py

import numpy as np
from sklearn.base import BaseEstimator, TransformerMixin
 
ROOMS, BEDROOMS, POPULATION, HOUSEHOLDS = 3, 4, 5, 6      # column indices
 
 
class AttributeAdder(BaseEstimator, TransformerMixin):
    """Add the three ratio features found during exploration."""
 
    def __init__(self, add_bedrooms_per_room=True):
        # No validation and no *args/**kwargs here — get_params relies on
        # every constructor argument being stored under its own name.
        self.add_bedrooms_per_room = add_bedrooms_per_room
 
    def fit(self, X, y=None):
        return self          # nothing to learn; still required by the API
 
    def transform(self, X):
        rooms_per_household = X[:, ROOMS] / X[:, HOUSEHOLDS]
        population_per_household = X[:, POPULATION] / X[:, HOUSEHOLDS]
        if self.add_bedrooms_per_room:
            bedrooms_per_room = X[:, BEDROOMS] / X[:, ROOMS]
            return np.c_[X, rooms_per_household, population_per_household,
                         bedrooms_per_room]
        return np.c_[X, rooms_per_household, population_per_household]
 
 
adder = AttributeAdder(add_bedrooms_per_room=False)
print(adder.get_params())     # {'add_bedrooms_per_room': False}

Because add_bedrooms_per_roomadd_bedrooms_per_room is a proper hyperparameter, a grid search can decide whether the feature helps:

python

GridSearchCV(full_pipeline, {"preparation__num__attribs__add_bedrooms_per_room": [True, False]})

python

GridSearchCV(full_pipeline, {"preparation__num__attribs__add_bedrooms_per_room": [True, False]})

For stateless transformations, FunctionTransformerFunctionTransformer skips the class entirely:

python

from sklearn.preprocessing import FunctionTransformer
log_transformer = FunctionTransformer(np.log1p, inverse_func=np.expm1, validate=True)

python

from sklearn.preprocessing import FunctionTransformer
log_transformer = FunctionTransformer(np.log1p, inverse_func=np.expm1, validate=True)

ColumnTransformer

Numeric and categorical columns need different treatment, and ColumnTransformerColumnTransformer routes them:

column_transformer.py

from sklearn.compose import ColumnTransformer
from sklearn.preprocessing import OneHotEncoder
 
num_attribs = list(housing_num.columns)          # 9 numeric columns
cat_attribs = ["ocean_proximity"]
 
full_preparation = ColumnTransformer([
    ("num", num_pipeline, num_attribs),
    ("cat", OneHotEncoder(handle_unknown="ignore"), cat_attribs),
])
 
housing_prepared = full_preparation.fit_transform(housing)
print(housing_prepared.shape)      # (16512, 17)  = 9 + 3 ratios + 5 categories

column_transformer.py

from sklearn.compose import ColumnTransformer
from sklearn.preprocessing import OneHotEncoder
 
num_attribs = list(housing_num.columns)          # 9 numeric columns
cat_attribs = ["ocean_proximity"]
 
full_preparation = ColumnTransformer([
    ("num", num_pipeline, num_attribs),
    ("cat", OneHotEncoder(handle_unknown="ignore"), cat_attribs),
])
 
housing_prepared = full_preparation.fit_transform(housing)
print(housing_prepared.shape)      # (16512, 17)  = 9 + 3 ratios + 5 categories

Each tuple is (name, transformer, columns)(name, transformer, columns). Anything not listed is dropped by default; remainder="passthrough"remainder="passthrough" keeps it instead. Output columns appear in the order the transformers are listed.

The routing is easier to see than to read:

diagram Diagram mermaid

  flowchart LR
  RAW["housing DataFrame
16,512 rows x 10 columns"] --> CT{{"ColumnTransformer"}}
  CT -->|"num: 9 numeric columns"| NP["num_pipeline"]
  CT -->|"cat: ocean_proximity"| OH["OneHotEncoder
handle_unknown='ignore'"]
  CT -.->|"not listed"| DROP["dropped
unless remainder='passthrough'"]
  NP --> S1["SimpleImputer
strategy='median'"]
  S1 --> S2["CombinedAttributesAdder
+3 ratio columns"]
  S2 --> S3["StandardScaler"]
  S3 --> OUT1["12 numeric columns"]
  OH --> OUT2["5 indicator columns"]
  OUT1 --> CAT["hstack in the order
the transformers were listed"]
  OUT2 --> CAT
  CAT --> FINAL["housing_prepared
16,512 x 17"]

Two properties are worth naming. The numeric branch is itself a PipelinePipeline, so nesting is the normal case rather than a trick. And the final width is not something you choose — it is $9 + 3 + 5 = 17$ , determined by what each branch emits, which is why get_feature_names_out()get_feature_names_out() exists.

Leakage, demonstrated

The argument for pipelines is not tidiness. It is that manual preprocessing leaks, and leakage is invisible in the metrics.

Two bars showing 5-fold cross-validated accuracy: 0.867 for feature selection performed before splitting, and 0.400 for selection inside a pipeline, with a dashed line at 0.50 marking the true accuracy. — There is no relationship in this data whatsoever, so the honest accuracy is 0.50. Selecting features on the full dataset before cross-validating manufactures 0.867 out of nothing.

leakage.py

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.normal(size=(120, 3000))       # pure noise
y = rng.integers(0, 2, 120)            # unrelated to X
 
# WRONG — selection sees every row, including the ones held out later
leaked = SelectKBest(f_classif, k=20).fit_transform(X, y)
print(cross_val_score(LogisticRegression(max_iter=2000), leaked, y, cv=5).mean())
# 0.8667
 
# RIGHT — selection happens inside each fold
honest = make_pipeline(SelectKBest(f_classif, k=20), LogisticRegression(max_iter=2000))
print(cross_val_score(honest, X, y, cv=5).mean())
# 0.4000

leakage.py

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.normal(size=(120, 3000))       # pure noise
y = rng.integers(0, 2, 120)            # unrelated to X
 
# WRONG — selection sees every row, including the ones held out later
leaked = SelectKBest(f_classif, k=20).fit_transform(X, y)
print(cross_val_score(LogisticRegression(max_iter=2000), leaked, y, cv=5).mean())
# 0.8667
 
# RIGHT — selection happens inside each fold
honest = make_pipeline(SelectKBest(f_classif, k=20), LogisticRegression(max_iter=2000))
print(cross_val_score(honest, X, y, cv=5).mean())
# 0.4000

Reading the plot

There is no signal. Both XX and yy are random and independent, so any honest estimate must sit near 0.50.
The leaky version reports 0.867. With 3,000 random columns, 20 of them correlate with yy by chance across the whole dataset — including the validation rows. The model then “predicts” those rows using columns chosen because they matched them.
The honest version reports 0.400. Below 0.50, because with 120 rows and 24 per fold the estimate is noisy. That noise is the truth; the 0.867 is not.
Nothing warns you. No error, no exception, no suspicious-looking output. On real data with real signal, leakage inflates a plausible 0.85 into a plausible 0.93, and you deploy it.

The general rule: any operation that learns from data must happen inside the cross-validation loop. That includes scaling, imputation, feature selection, target encoding, PCA, and oversampling. PipelinePipeline is what puts them there.

See it move

The sketch runs that same experiment live: pure-noise columns, a random target, five folds. Watch where the selector looks. On the left it is allowed to score every row before the split — including the fold that is about to be held out — and the accuracy it reports drifts upward as you give it more columns to choose from. On the right the selector is inside the fold, so more columns buy it nothing.

sketch Where the selector looks decides the score p5.js

Feature selection on random noise, run inside and outside the cross-validation loop as the number of candidate columns grows. The leaky score climbs with the number of columns; the honest score stays flat around chance level. Click to redraw the noise.

The red trace is not a bug in the classifier — it is a correct measurement of the wrong quantity. Give the selector 600 columns to sift and it will find four that happen to separate the held-out rows, because it was shown them.

The same procedure ported to NumPy and averaged over 40 random datasets, to check that the sketch is not a visual accident:

Candidate columns	Selected before splitting	Selected inside each fold
20	0.654	0.500
60	0.715	0.496
150	0.770	0.510
300	0.783	0.493
600	0.796	0.521

The honest column is flat at chance, as it must be on noise. The leaky column climbs monotonically with the number of candidates. The gap between the two is the size of the lie, and it widens in exactly the direction modern feature engineering pushes: more columns, more transformations, more chances for something to match the validation rows by accident.

Worked example by hand: the leaky scaler

Training values $[1, 2, 3]$ and one test value $[100]$ .

Wrong — fit on everything. Mean of all four values is 26.5, population standard deviation 42.4411:

\text{train} \to \left[\tfrac{1-26.5}{42.44},\; \tfrac{2-26.5}{42.44},\; \tfrac{3-26.5}{42.44}\right] = [-0.6008,\; -0.5773,\; -0.5537]

\text{test} \to \tfrac{100-26.5}{42.44} = 1.7318

Right — fit on training only. Mean 2, standard deviation 0.8165:

\text{train} \to [-1.2247,\; 0,\; +1.2247] \qquad \text{test} \to \tfrac{100-2}{0.8165} = 120.025

The two are not slightly different, they are different in kind. Fitted on everything, the training data is squeezed into a span of 0.047 and the test point looks unremarkable at 1.73. Fitted correctly, the training data spans 2.45 and the test point is revealed for what it is: an extreme value 120 standard deviations out.

The leaky version does not merely produce a nicer number. It hides the fact that the test point is an outlier — which is exactly the information the model most needed.

The full pipeline

full_pipeline.py

from sklearn.compose import ColumnTransformer
from sklearn.ensemble import RandomForestRegressor
from sklearn.impute import SimpleImputer
from sklearn.model_selection import cross_val_score
from sklearn.pipeline import Pipeline
from sklearn.preprocessing import OneHotEncoder, StandardScaler
 
num_pipeline = Pipeline([
    ("imputer", SimpleImputer(strategy="median")),
    ("attribs", AttributeAdder()),
    ("scaler", StandardScaler()),
])
 
preparation = ColumnTransformer([
    ("num", num_pipeline, num_attribs),
    ("cat", OneHotEncoder(handle_unknown="ignore"), cat_attribs),
])
 
full_pipeline = Pipeline([
    ("preparation", preparation),
    ("model", RandomForestRegressor(n_estimators=80, random_state=42, n_jobs=-1)),
])
 
# One call. Preprocessing is refitted inside every fold, automatically.
scores = -cross_val_score(full_pipeline, housing, housing_labels, cv=5,
                          scoring="neg_root_mean_squared_error")
print(f"RMSE {scores.mean():,.0f} (+/- {scores.std():,.0f})")
 
full_pipeline.fit(housing, housing_labels)
predictions = full_pipeline.predict(new_districts)     # raw DataFrame in, prices out

full_pipeline.py

from sklearn.compose import ColumnTransformer
from sklearn.ensemble import RandomForestRegressor
from sklearn.impute import SimpleImputer
from sklearn.model_selection import cross_val_score
from sklearn.pipeline import Pipeline
from sklearn.preprocessing import OneHotEncoder, StandardScaler
 
num_pipeline = Pipeline([
    ("imputer", SimpleImputer(strategy="median")),
    ("attribs", AttributeAdder()),
    ("scaler", StandardScaler()),
])
 
preparation = ColumnTransformer([
    ("num", num_pipeline, num_attribs),
    ("cat", OneHotEncoder(handle_unknown="ignore"), cat_attribs),
])
 
full_pipeline = Pipeline([
    ("preparation", preparation),
    ("model", RandomForestRegressor(n_estimators=80, random_state=42, n_jobs=-1)),
])
 
# One call. Preprocessing is refitted inside every fold, automatically.
scores = -cross_val_score(full_pipeline, housing, housing_labels, cv=5,
                          scoring="neg_root_mean_squared_error")
print(f"RMSE {scores.mean():,.0f} (+/- {scores.std():,.0f})")
 
full_pipeline.fit(housing, housing_labels)
predictions = full_pipeline.predict(new_districts)     # raw DataFrame in, prices out

Three properties follow, and each one is the point:

Cross-validation is honest. Every fold refits the imputer, the scaler and the encoder on that fold’s training rows only.
Deployment is one object. joblib.dump(full_pipeline, "model.pkl")joblib.dump(full_pipeline, "model.pkl") saves the preprocessing with the model. There is no second script to keep in sync.
Tuning reaches everything. GridSearchCVGridSearchCV can search preparation__num__imputer__strategypreparation__num__imputer__strategy alongside model__n_estimatorsmodel__n_estimators, using the double-underscore path through the nesting.

Pitfalls

quizCheck yourself

Feature selection on 3,000 random columns before cross-validation reported 0.867 accuracy on random labels. Why?
Show answer
B — Some columns correlate with the target by chance across the whole dataset, including the validation rows — so the model is scored on rows that helped choose its features — With 3,000 columns and 120 rows, chance correlations are guaranteed. Selecting on all the data lets validation rows influence which columns survive, and the score becomes self-fulfilling.
What is the rule that makes a pipeline leak-proof?
Show answer
B — Any operation that learns parameters from data must run inside the cross-validation loop, which is exactly what Pipeline arranges — Scaling, imputation, selection, encoding and PCA all learn from data. Fitted outside the loop, each one leaks; inside a Pipeline, cross_val_score refits them per fold automatically.
Why must a custom transformer store its constructor arguments unchanged, under the same names?
Show answer
B — Because get_params reads the constructor signature and looks up matching attributes — renaming or coercing them breaks cloning inside cross-validation and grid search — BaseEstimator's get_params introspects __init__ and reads self.<argname>. If that attribute is missing or altered, clone() fails and every CV fold errors out.
Scaling [1, 2, 3] and [100] together gives the training data a span of 0.047 and puts the test point at 1.73. Fitted on training only, the test point lands at 120. What is the real cost of the leaky version?
Show answer
B — It hides that the test point is an extreme outlier — precisely the information the model needed — The leaky scaler absorbs the outlier into its own statistics, so the outlier stops looking like one. Honest scaling exposes it at 120 standard deviations out.

🧪 Try It Yourself

Exercise 1 – Chain steps with Pipeline

Exercise 2 – Write a custom transformer

Exercise 3 – Route columns with ColumnTransformer

Exercise 4 – Manufacture accuracy from noise

Exercise 5 – The leaky scaler, by hand

Recap

Manual preprocessing means keeping several fitted objects in sync, and it cannot be cross-validated honestly.
PipelinePipeline chains transformers into one estimator; ColumnTransformerColumnTransformer routes different columns to different treatments.
A custom transformer needs fitfit returning selfself and transformtransform; BaseEstimatorBaseEstimator and TransformerMixinTransformerMixin supply the rest — provided constructor arguments are stored unchanged.
Leakage demonstrated: 3,000 random columns, a random target, and selecting before splitting reports 0.867 against a true 0.50.
The leaky-scaler example does more than distort a number: it hides the test outlier at 1.73 instead of exposing it at 120.
Any operation that learns from data belongs inside the cross-validation loop. PipelinePipeline is the mechanism that puts it there.

Exercise 6 – Watch the lie grow with the number of columns

Continue to End-to-End Machine Learning Project (California Housing) — assemble everything in this phase, train three models, tune the winner and open the test set exactly once.

If this helped you, consider buying me a coffee ☕

Buy me a coffee

Transformation Pipelines & Custom Transformers

What you’ll learn

Intuition

The problem with loose steps

Pipeline

A custom transformer

ColumnTransformer

Leakage, demonstrated

Reading the plot

See it move

Worked example by hand: the leaky scaler

The full pipeline

Pitfalls

🧪 Try It Yourself

Exercise 1 – Chain steps with Pipeline

Exercise 2 – Write a custom transformer

Exercise 3 – Route columns with ColumnTransformer

Exercise 4 – Manufacture accuracy from noise

Exercise 5 – The leaky scaler, by hand

Recap

Exercise 6 – Watch the lie grow with the number of columns

Next

Was this page helpful?