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

> Padroneggiare le pipeline scikit-learn con ColumnTransformer per il feature engineering. Best practice per il preprocessing e preparazione ai colloqui ML.

- Published: 2026-09-02
- Updated: 2026-09-02
- Author: Anthony Fillion-Maillet
- Reading time: 5 min

---

Le pipeline di scikit-learn trasformano codice di preprocessing caotico in workflow riproducibili e pronti per la produzione. Con la [versione 1.9](https://scikit-learn.org/stable/whats_new/v1.9.html) 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.

> **Suggerimento per il Colloquio**
>
> Gli intervistatori spesso chiedono: "Come si previene il data leakage durante la cross-validation?" La risposta sono le pipeline. Addestrare i passaggi di preprocessing all'interno del ciclo CV assicura che i dati di test non influenzino mai le decisioni di scaling o encoding.

## 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:

```python
# 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:

```python
# 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:

```python
# 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`:

```python
# 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`:

```python
# 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:

```python
# 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](https://scikit-learn.org/stable/whats_new/v1.9.html) aggiunge il monitoraggio del progresso alle pipeline:

```python
# 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:

```python
# 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:

```python
# 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.

> **Compatibilità delle Versioni**
>
> Le pipeline serializzate con scikit-learn 1.8 potrebbero non caricarsi correttamente sulla 1.9 se utilizzano parametri deprecati. La versione di scikit-learn dovrebbe essere sempre fissata in produzione e gli upgrade testati prima del deployment.

## 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

---

Source: SharpSkill (https://sharpskill.dev), tech interview preparation for your real stack.
HTML version of this page: https://sharpskill.dev/it/blog/data-science/scikit-learn-pipelines-feature-engineering-interview-2026