Scikit-Learn Pipelines nel 2026: Feature Engineering e Domande da Colloquio

Le pipeline di scikit-learn trasformano codice di preprocessing caotico in workflow riproducibili e pronti per la produzione. Con la versione 1.9 rilasciata a giugno 2026, Pipeline e ColumnTransformer rimangono il fondamento di qualsiasi progetto di machine learning serio, ora con monitoraggio tramite callback, visualizzazione HTML migliorata e supporto Array API.

Perché le Pipeline eliminano il Data Leakage

Il data leakage si verifica quando le informazioni dal test set influenzano il preprocessing. Un errore comune: addestrare lo StandardScaler sull'intero dataset prima di dividere. Lo scaler impara media e varianza dai campioni di test, gonfiando artificialmente i punteggi di cross-validation.

Le pipeline risolvono questo problema concatenando transformer ed estimator in un singolo oggetto che addestra tutti i passaggi insieme:

# pipeline_basics.py
from sklearn.pipeline import Pipeline
from sklearn.preprocessing import StandardScaler
from sklearn.linear_model import LogisticRegression
from sklearn.model_selection import cross_val_score

# Create a pipeline that scales features and trains a classifier
pipeline = Pipeline([
    ('scaler', StandardScaler()),  # Step 1: Normalize features
    ('classifier', LogisticRegression(max_iter=1000))  # Step 2: Train model
])

# cross_val_score fits the scaler separately for each fold
# No data leakage: test fold is never seen during scaler.fit()
scores = cross_val_score(pipeline, X, y, cv=5, scoring='accuracy')
print(f"Mean accuracy: {scores.mean():.3f} (+/- {scores.std():.3f})")

L'oggetto Pipeline implementa fit, predict e score, rendendolo intercambiabile con qualsiasi estimator di scikit-learn. Questo significa che GridSearchCV, RandomizedSearchCV e tutte le utility di cross-validation funzionano senza modifiche.

ColumnTransformer per Tipi di Dati Misti

I dataset reali contengono colonne numeriche (età, stipendio) e colonne categoriche (paese, tipo_prodotto). ColumnTransformer applica preprocessing differenti a diversi sottoinsiemi di colonne, poi concatena i risultati:

# column_transformer_example.py
from sklearn.compose import ColumnTransformer
from sklearn.preprocessing import StandardScaler, OneHotEncoder
from sklearn.impute import SimpleImputer
from sklearn.pipeline import Pipeline
import pandas as pd

# Sample dataset with mixed types
df = pd.DataFrame({
    'age': [25, 30, None, 45],
    'salary': [50000, 60000, 75000, None],
    'country': ['US', 'UK', 'DE', 'US'],
    'education': ['Bachelor', 'Master', 'PhD', 'Bachelor']
})

# Define numeric preprocessing: impute missing values, then scale
numeric_transformer = Pipeline([
    ('imputer', SimpleImputer(strategy='median')),
    ('scaler', StandardScaler())
])

# Define categorical preprocessing: impute with mode, then one-hot encode
categorical_transformer = Pipeline([
    ('imputer', SimpleImputer(strategy='most_frequent')),
    ('encoder', OneHotEncoder(handle_unknown='ignore', sparse_output=False))
])

# Combine transformers with ColumnTransformer
preprocessor = ColumnTransformer(
    transformers=[
        ('num', numeric_transformer, ['age', 'salary']),
        ('cat', categorical_transformer, ['country', 'education'])
    ],
    verbose_feature_names_out=True  # Prefix output names with transformer name
)

# Fit and inspect output shape
X_transformed = preprocessor.fit_transform(df)
print(f"Transformed shape: {X_transformed.shape}")
print(f"Feature names: {preprocessor.get_feature_names_out()}")

Il parametro verbose_feature_names_out (migliorato nella versione 1.6 per accettare stringhe e callable) controlla la denominazione delle feature in output. Impostandolo su True si aggiunge il nome del transformer come prefisso a ogni feature, prevenendo collisioni di nomi quando più transformer generano nomi di colonna simili.

Selezione Automatica delle Colonne con make_column_selector

I nomi di colonna hardcoded falliscono quando i dataset cambiano. make_column_selector seleziona le colonne automaticamente per dtype:

# automatic_column_selection.py
from sklearn.compose import ColumnTransformer, make_column_selector
from sklearn.preprocessing import StandardScaler, OneHotEncoder

# Select columns by dtype: no hardcoded column names
preprocessor = ColumnTransformer(
    transformers=[
        ('num', StandardScaler(), make_column_selector(dtype_include='number')),
        ('cat', OneHotEncoder(handle_unknown='ignore'),
         make_column_selector(dtype_include='object'))
    ],
    remainder='passthrough'  # Keep unprocessed columns as-is
)

# Works with any DataFrame that has numeric and object columns
X_processed = preprocessor.fit_transform(df)

Il parametro remainder controlla cosa succede alle colonne non abbinate da alcun transformer. Le opzioni includono 'drop' (default), 'passthrough' (mantieni invariate) o un transformer da applicare alle colonne rimanenti.

Feature Engineering all'interno delle Pipeline

Le pipeline si estendono oltre il preprocessing per includere passaggi di feature engineering. I transformer personalizzati ereditano da BaseEstimator e TransformerMixin:

# custom_transformer.py
from sklearn.base import BaseEstimator, TransformerMixin
import numpy as np

class DateFeatureExtractor(BaseEstimator, TransformerMixin):
    """Extract year, month, day, and weekday from datetime columns."""
    
    def __init__(self, date_column='date'):
        self.date_column = date_column
    
    def fit(self, X, y=None):
        # No fitting required for date extraction
        return self
    
    def transform(self, X):
        X = X.copy()
        dates = pd.to_datetime(X[self.date_column])
        
        # Extract temporal features
        X['year'] = dates.dt.year
        X['month'] = dates.dt.month
        X['day'] = dates.dt.day
        X['weekday'] = dates.dt.weekday
        X['is_weekend'] = (dates.dt.weekday >= 5).astype(int)
        
        # Drop original date column
        return X.drop(columns=[self.date_column])
    
    def get_feature_names_out(self, input_features=None):
        # Required for Pipeline.get_feature_names_out() to work
        return ['year', 'month', 'day', 'weekday', 'is_weekend']

# Integrate into a full pipeline
full_pipeline = Pipeline([
    ('date_features', DateFeatureExtractor(date_column='signup_date')),
    ('preprocessor', preprocessor),
    ('model', LogisticRegression())
])

Implementare get_feature_names_out permette alla nuova rappresentazione HTML di scikit-learn 1.9 di visualizzare i nomi delle feature in output, facilitando debugging e documentazione.

Tuning degli Iperparametri attraverso i Passaggi della Pipeline

GridSearchCV effettua il tuning dei parametri attraverso tutti i passaggi della pipeline usando la convenzione di denominazione step__parameter:

# pipeline_gridsearch.py
from sklearn.model_selection import GridSearchCV
from sklearn.ensemble import RandomForestClassifier

# Build pipeline with tunable components
pipeline = Pipeline([
    ('preprocessor', preprocessor),
    ('classifier', RandomForestClassifier(random_state=42))
])

# Define parameter grid: note the double underscore syntax
param_grid = {
    'preprocessor__num__imputer__strategy': ['mean', 'median'],
    'classifier__n_estimators': [100, 200],
    'classifier__max_depth': [10, 20, None]
}

# GridSearchCV handles all combinations automatically
grid_search = GridSearchCV(
    pipeline,
    param_grid,
    cv=5,
    scoring='f1_weighted',
    n_jobs=-1
)

grid_search.fit(X_train, y_train)
print(f"Best params: {grid_search.best_params_}")
print(f"Best score: {grid_search.best_score_:.3f}")

Il doppio underscore (__) accede ai parametri annidati. Per ColumnTransformer, il percorso include il nome del transformer: preprocessor__num__imputer__strategy raggiunge il parametro strategy dell'imputer all'interno del transformer numerico.

Caching dei Transformer con il Parametro memory

Il preprocessing complesso su grandi dataset può essere lento. Il parametro memory memorizza nella cache i transformer addestrati su disco:

# cached_pipeline.py
from sklearn.pipeline import Pipeline
from tempfile import mkdtemp
import joblib

# Create a cache directory
cachedir = mkdtemp()

# Pipeline caches intermediate steps
cached_pipeline = Pipeline(
    [
        ('preprocessor', preprocessor),  # Cached after first fit
        ('classifier', RandomForestClassifier())
    ],
    memory=cachedir  # Or use joblib.Memory for more control
)

# First fit: slow (computes all transformations)
cached_pipeline.fit(X_train, y_train)

# Second fit with same X_train: fast (loads cached transformers)
cached_pipeline.fit(X_train, y_train)  # Skips preprocessor fitting

Il caching è particolarmente utile durante la ricerca degli iperparametri, dove lo stesso preprocessing viene applicato a più configurazioni del modello. L'estimator finale (classificatore o regressore) non viene mai memorizzato nella cache, solo i transformer intermedi.

Monitoraggio del Training con Callback (scikit-learn 1.9)

L'API callback nella versione 1.9 aggiunge il monitoraggio del progresso alle pipeline:

# callback_monitoring.py
from sklearn.callback import ProgressBar, ScoringMonitor
from sklearn.linear_model import LogisticRegression

# Display progress during GridSearchCV
with ProgressBar():
    grid_search = GridSearchCV(
        pipeline,
        param_grid,
        cv=5,
        n_jobs=1  # Progress bars require single-threaded execution
    )
    grid_search.fit(X_train, y_train)

# Monitor scoring metrics per iteration
with ScoringMonitor(X_val, y_val, scoring='accuracy') as monitor:
    logreg = LogisticRegression(solver='lbfgs', max_iter=500)
    logreg.fit(X_train, y_train)
    print(f"Scores per iteration: {monitor.scores_}")

I callback funzionano con Pipeline, StandardScaler, LogisticRegression (solver LBFGS) e tutte le classi Search CV. Questa funzionalità sperimentale fornisce visibilità sui fit di lunga durata senza codice di logging personalizzato.

FeatureUnion per Estrazione Parallela di Feature

FeatureUnion esegue più transformer in parallelo e concatena il loro output orizzontalmente:

# feature_union_example.py
from sklearn.pipeline import FeatureUnion, Pipeline
from sklearn.decomposition import PCA
from sklearn.feature_selection import SelectKBest, f_classif

# Combine PCA and univariate selection
feature_union = FeatureUnion([
    ('pca', PCA(n_components=5)),
    ('select', SelectKBest(f_classif, k=10))
])

# Pipeline: preprocess, combine features, classify
pipeline = Pipeline([
    ('preprocessor', preprocessor),
    ('features', feature_union),
    ('classifier', LogisticRegression())
])

# Output contains 5 PCA components + 10 selected features = 15 features
pipeline.fit(X_train, y_train)

La versione 1.7.2 ha aggiunto una validazione che assicura che tutti i transformer restituiscano output 2D. Questo rileva errori precocemente quando un transformer personalizzato restituisce accidentalmente un array 1D.

Serializzazione e Deployment

Le pipeline vengono serializzate con joblib, preservando tutti i parametri addestrati:

# model_deployment.py
import joblib

# Save the entire pipeline (preprocessing + model)
joblib.dump(pipeline, 'churn_predictor.pkl')

# Load in production
loaded_pipeline = joblib.load('churn_predictor.pkl')

# Predict on new data (same preprocessing automatically applied)
new_data = pd.DataFrame({'age': [32], 'salary': [65000], ...})
predictions = loaded_pipeline.predict(new_data)

Il file serializzato contiene tutto: logica di selezione delle colonne, medie e varianze dello scaler addestrato, categorie dell'encoder e pesi del modello. Il deployment richiede solo scikit-learn e joblib, nessun codice di preprocessing personalizzato.

Domande da Colloquio sulle Pipeline Scikit-Learn

D: Cos'è il data leakage e come lo prevengono le pipeline?

Il data leakage si verifica quando informazioni dal validation o test set influenzano l'addestramento del modello. L'esempio classico: addestrare uno StandardScaler sull'intero dataset prima di dividere. La media e la deviazione standard dello scaler includono campioni di test, gonfiando artificialmente i punteggi di cross-validation.

Le pipeline prevengono questo incapsulando il preprocessing all'interno del ciclo di cross-validation. Quando cross_val_score chiama pipeline.fit(), ogni fold addestra lo scaler solo sui dati di training. Le statistiche del test fold non penetrano mai nel preprocessing.

D: Come si accede e si modificano i parametri dei passaggi annidati della pipeline?

Si utilizza la sintassi del doppio underscore: pipeline.set_params(classifier__n_estimators=200). Per ColumnTransformer, si include il nome del transformer: pipeline.set_params(preprocessor__num__scaler__with_mean=False). Il metodo get_params() restituisce tutti i parametri annidati come dizionario piatto, utile per ispezione e logging.

D: Quando si dovrebbe usare FeatureUnion rispetto a ColumnTransformer?

ColumnTransformer applica transformer diversi a colonne diverse dello stesso input. FeatureUnion applica transformer diversi all'intero input e concatena orizzontalmente. Si usa ColumnTransformer per il preprocessing eterogeneo delle colonne (numerico vs categorico). Si usa FeatureUnion per l'augmentation delle feature, dove metodi di estrazione multipli (PCA, feature polinomiali, estrattori specifici del dominio) producono feature complementari dagli stessi dati.

D: Come si fa il debug di una pipeline che fallisce?

Impostare verbose=True sulla pipeline per vedere i tempi dei passaggi. Accedere ai risultati intermedi con pipeline.named_steps['step_name'].transform(X) per una pipeline addestrata. In scikit-learn 1.9, la rappresentazione HTML visualizza gli attributi addestrati e i nomi delle feature in output, semplificando il debugging nei notebook Jupyter. Per un'ispezione più profonda, usare pipeline[:-1].fit_transform(X, y) per ottenere l'output subito prima dell'estimator finale.

Checklist di Produzione per Pipeline ML

• Fissare la versione di scikit-learn nei requirements per evitare problemi di serializzazione
• Usare make_column_selector invece di nomi di colonna hardcoded quando i dataset possono cambiare
• Impostare handle_unknown='ignore' su OneHotEncoder per gestire nuove categorie in produzione
• Memorizzare nella cache il preprocessing costoso durante lo sviluppo con il parametro memory
• Validare la shape dell'output della pipeline e i nomi delle feature nei test prima del deployment
• Salvare get_feature_names_out() insieme al modello per l'analisi dell'importanza delle feature
• Usare Pipeline all'interno di GridSearchCV, mai il contrario, per assicurare il corretto comportamento CV