DEV Community

Akshay Dev Karama
Akshay Dev Karama

Posted on

Metadata Routing

#python #tutorial #machinelearning #datascience

Stop Fighting Scikit-Learn Pipelines: How Metadata Routing Fixes Sample Weights & Groups

A couple of months ago, I stumbled upon this video by Vincent D. Warmerdam about metadata routing in scikit-learn. I'll be honest, I had no idea what "metadata routing" even meant, but Vincent's explanation completely changed how I think about building ML pipelines.

The video showed me that one of the most frustrating problems in scikit-learn; passing sample weights and groups through complex pipelines finally had an elegant solution. It piqued my curiosity enough that I dove deep into the feature, tested it extensively, and honestly, I was surprised by how little coverage this gets in technical blogs and articles. So I figured, why not write about it myself and share what I learned?

If you've ever struggled with imbalanced datasets, grouped cross-validation, or just wanted to pass custom information through your pipelines, this article is for you. Let's start from the very beginning.

What is "Metadata" in Machine Learning?

Let's start with a concrete example. You're building a credit card fraud detection model with this data:

# Your training data
X = transaction_features  # Amount, merchant, time, location, etc.
y = is_fraud             # 0 = legitimate, 1 = fraud

# But you also have additional information:
sample_weights = [1.0, 1.0, 10.0, 1.0, ...]  # Fraud transactions weighted 10x
customer_ids = [101, 102, 101, 103, ...]      # Which customer made each transaction
Enter fullscreen mode Exit fullscreen mode

Metadata is the "extra information" beyond your features (X) and labels (y):

  • sample_weight: How important is each transaction? (Fraud = 10x more important)
  • groups: Which customer does each transaction belong to? (For proper cross-validation)
  • Custom metadata: Transaction timestamps, confidence scores, data quality flags, etc.

Why Metadata Matters: The Credit Card Fraud Problem

Imagine you're building a fraud detection system for a financial company. You have:

  • Imbalanced data: 99% legitimate transactions, 1% fraudulent
  • Time-series data: Transactions grouped by customer ID (can't split customers across train/test)
  • Complex pipeline: Feature scaling → feature selection → classification
  • Business requirement: False negatives (missed fraud) cost 10x more than false positives

The Challenge: Your model needs to:

  1. Weight samples - Treat fraudulent transactions as 10x more important during training
  2. Respect customer grouping - Keep all transactions from the same customer together during cross-validation (otherwise you're leaking information!)
  3. Pass this information through pipelines - Your scaler, feature selector, and classifier all need access to these weights
  4. Work with hyperparameter tuning - GridSearchCV needs to use both weights and groups

The problem?

This "metadata" (weights, groups) isn't part of your feature matrix X or labels y. It's auxiliary information that needs to flow through your entire ML pipeline.

Before scikit-learn 1.3, this was nearly impossible. Let's see why.

The Challenge: Why Metadata Routing Was Needed

Prior to metadata routing, you'd face multiple interconnected problems:

Problem 1: Can't Pass Weights Through Pipelines 

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

# Your fraud detection pipeline
pipe = Pipeline([
    ('scaler', StandardScaler()),
    ('clf', LogisticRegression())
])

# You have fraud weights (fraudulent transactions weighted 10x)
fraud_weights = np.where(y == 1, 10.0, 1.0)

# This doesn't work!
pipe.fit(X, y, sample_weight=fraud_weights)  # Error: unexpected keyword argument
Enter fullscreen mode Exit fullscreen mode

Problem 2: Can't Use Groups in Cross-Validation 

from sklearn.model_selection import cross_val_score, GroupKFold

# You have customer IDs (can't split customers across folds)
customer_groups = df['customer_id'].values

# This doesn't work with pipelines!
scores = cross_val_score(
    pipe, X, y,
    cv=GroupKFold(n_splits=5),
    groups=customer_groups  # Pipeline doesn't know what to do with this
)
Enter fullscreen mode Exit fullscreen mode

Problem 3: Can't Combine Both in GridSearchCV 

from sklearn.model_selection import GridSearchCV

# You need BOTH weights AND groups during hyperparameter tuning
grid = GridSearchCV(pipe, param_grid, cv=GroupKFold(n_splits=5))

# This is impossible - can't pass both!
grid.fit(X, y, sample_weight=fraud_weights, groups=customer_groups)  # Doesn't work
Enter fullscreen mode Exit fullscreen mode

So you can begin to see the problem by now. Pipelines had no way to route this metadata to specific components. You'd have to use hacky workarounds like clf__sample_weight, which was inconsistent, broke with nested pipelines, and completely failed with cross-validation.

The Solution: Metadata Routing API

Metadata routing solves ALL three problems at once with a clean, explicit API. Here's how it transforms our fraud detection pipeline:

from sklearn import set_config
from sklearn.model_selection import GridSearchCV, GroupKFold

# Enable metadata routing globally
set_config(enable_metadata_routing=True)

# Build the fraud detection pipeline
pipe = Pipeline([
    ('scaler', StandardScaler()),
    ('clf', LogisticRegression())
])

# Configure metadata routing - declare what each component needs
pipe['clf'].set_fit_request(sample_weight=True)
pipe['clf'].set_score_request(sample_weight=True)

# Problem 1 SOLVED: Pass weights through pipeline
pipe.fit(X, y, sample_weight=fraud_weights)

# Problem 2 SOLVED: Use groups in cross-validation
scores = cross_val_score(
    pipe, X, y,
    cv=GroupKFold(n_splits=5),
    groups=customer_groups  # Works perfectly!
)

# Problem 3 SOLVED: Combine weights AND groups in GridSearchCV
grid = GridSearchCV(pipe, param_grid, cv=GroupKFold(n_splits=5))
grid.fit(X, y, sample_weight=fraud_weights, groups=customer_groups)  # Both work!

print(f"Best model handles imbalance AND respects customer grouping!")
Enter fullscreen mode Exit fullscreen mode

What changed? Each component explicitly declares what metadata it needs using set_*_request() methods. The pipeline then automatically routes metadata to the right places. Simple, explicit, powerful.

Here's what you need to know:

  • set_fit_request(): Declares metadata needed during fit()
  • set_score_request(): Declares metadata needed during score()
  • set_predict_request(): Declares metadata needed during predict()
  • Explicit routing: You must explicitly declare what metadata each component receives
  • Selective propagation: Metadata is only routed to components that request it - not all components automatically receive it

Important:
The pipeline doesn't pass metadata to every step. Only components that explicitly call set_*_request(metadata=True) will receive that metadata. Components that don't request metadata won't receive it, even if you pass it to the pipeline.

# Example: Selective routing
pipe = Pipeline([
    ('scaler', StandardScaler()),        # Doesn't request sample_weight
    ('clf', LogisticRegression())        # Requests sample_weight
])

pipe['clf'].set_fit_request(sample_weight=True)  # Only clf gets weights

# When you call:
pipe.fit(X, y, sample_weight=weights)

# What happens:
# - scaler.fit(X, y) → NO sample_weight (didn't request it)
# - clf.fit(X_scaled, y, sample_weight=weights) → Gets sample_weight (requested it)
Enter fullscreen mode Exit fullscreen mode

Implementation Guide

Example 1: Custom Transformer with Metadata

Let's build a custom transformer that uses sample weights during fitting. This is useful for weighted feature scaling or selection.

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

class WeightedStandardScaler(BaseEstimator, TransformerMixin):
    """StandardScaler that respects sample weights during fitting."""

    def __init__(self):
        self.mean_ = None
        self.std_ = None

    def fit(self, X, y=None, sample_weight=None):
        """Fit scaler using weighted mean and std."""
        if sample_weight is None:
            sample_weight = np.ones(X.shape[0])

        # Normalize weights
        sample_weight = sample_weight / sample_weight.sum()

        # Compute weighted statistics
        self.mean_ = np.average(X, axis=0, weights=sample_weight)
        variance = np.average((X - self.mean_) ** 2, axis=0, weights=sample_weight)
        self.std_ = np.sqrt(variance)

        return self

    def transform(self, X):
        """Transform using fitted statistics."""
        return (X - self.mean_) / self.std_

    def get_metadata_routing(self):
        """Configure metadata routing for this transformer."""
        return (
            super()
            .get_metadata_routing()
            .add_self_request(self)
            .fit(sample_weight=True)  # Request sample_weight in fit()
        )

# Usage
from sklearn import set_config
set_config(enable_metadata_routing=True)

X = np.random.randn(100, 5)
weights = np.random.rand(100)

scaler = WeightedStandardScaler()
X_scaled = scaler.fit_transform(X, sample_weight=weights)
Enter fullscreen mode Exit fullscreen mode

Here's what matters when building custom estimators:

  1. Accept sample_weight parameter in fit() method
  2. Implement get_metadata_routing() to declare routing requirements
  3. Use add_self_request() and chain routing configuration
  4. Handle None case when metadata isn't provided

Example 2: Pipeline with Metadata Routing

Now let's use our custom transformer in a pipeline with multiple metadata consumers.

from sklearn.pipeline import Pipeline
from sklearn.linear_model import LogisticRegression
from sklearn.model_selection import train_test_split

# Create sample data
X = np.random.randn(1000, 10)
y = (X[:, 0] + X[:, 1] > 0).astype(int)
sample_weights = np.random.rand(1000)

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

# Build pipeline with metadata routing
pipe = Pipeline([
    ('scaler', WeightedStandardScaler()),
    ('classifier', LogisticRegression(max_iter=1000))
])

# Configure routing: both steps need sample_weight
pipe.set_fit_request(sample_weight=True)
pipe['classifier'].set_fit_request(sample_weight=True)

# Fit with sample weights - they're routed to both steps
pipe.fit(X_train, y_train, sample_weight=w_train)

# Score also supports metadata routing
pipe['classifier'].set_score_request(sample_weight=True)
score = pipe.score(X_test, y_test, sample_weight=w_test)

print(f"Weighted accuracy: {score:.3f}")
Enter fullscreen mode Exit fullscreen mode

Pipeline Routing Rules:

  • Each step must explicitly request metadata via set_*_request()
  • The pipeline itself can also request metadata to pass through
  • Metadata is only routed to steps that request it
  • You can route different metadata to different steps

Example 3: GridSearchCV with Metadata

Metadata routing shines in hyperparameter tuning scenarios where you need to pass weights or groups to cross-validation.

from sklearn.model_selection import GridSearchCV
from sklearn.datasets import make_classification

# Generate imbalanced dataset
X, y = make_classification(
    n_samples=1000, n_features=20, n_informative=15,
    n_redundant=5, weights=[0.9, 0.1], random_state=42
)

# Create sample weights to handle imbalance
sample_weights = np.where(y == 1, 10.0, 1.0)

# Build pipeline
pipe = Pipeline([
    ('scaler', WeightedStandardScaler()),
    ('clf', LogisticRegression(max_iter=1000))
])

# Configure metadata routing for both steps
pipe['scaler'].set_fit_request(sample_weight=True)
pipe['clf'].set_fit_request(sample_weight=True)
pipe['clf'].set_score_request(sample_weight=True)

# GridSearchCV with metadata routing
param_grid = {
    'clf__C': [0.1, 1.0, 10.0],
    'clf__penalty': ['l1', 'l2']
}

grid_search = GridSearchCV(
    pipe,
    param_grid,
    cv=5,
    scoring='accuracy',
    n_jobs=-1
)

# Fit with sample weights - they're used in both fitting and scoring
grid_search.fit(X, y, sample_weight=sample_weights)

print(f"Best params: {grid_search.best_params_}")
print(f"Best weighted score: {grid_search.best_score_:.3f}")

# Access the best model
best_pipe = grid_search.best_estimator_
Enter fullscreen mode Exit fullscreen mode

GridSearchCV Routing Features:

  • Metadata is automatically passed to all CV folds
  • Both fitting and scoring can use metadata
  • Works with custom scorers that accept metadata
  • Supports groups parameter for GroupKFold and similar splitters