Pipelines Scikit-Learn em 2026: Feature Engineering e Perguntas de Entrevista

Os pipelines do scikit-learn transformam código de pré-processamento caótico em workflows reproduzíveis e prontos para produção. Com a versão 1.9 lançada em junho de 2026, Pipeline e ColumnTransformer continuam sendo a base de qualquer projeto sério de machine learning, agora com monitoramento via callbacks, visualização HTML aprimorada e suporte a Array API.

Por Que os Pipelines Eliminam o Vazamento de Dados

O vazamento de dados ocorre quando informações do conjunto de teste influenciam o pré-processamento. Um erro comum: ajustar um StandardScaler em todo o dataset antes de dividi-lo. O scaler aprende a média e variância das amostras de teste, inflando artificialmente os scores de validação cruzada.

Os pipelines corrigem isso encadeando transformers e estimadores em um único objeto que ajusta todas as etapas juntas:

# 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})")

O objeto Pipeline implementa fit, predict e score, tornando-o intercambiável com qualquer estimador do scikit-learn. Isso significa que GridSearchCV, RandomizedSearchCV e todos os utilitários de validação cruzada funcionam sem modificação.

ColumnTransformer para Tipos de Dados Mistos

Datasets reais contêm colunas numéricas (idade, salário) e colunas categóricas (país, tipo_produto). ColumnTransformer aplica diferentes pré-processamentos a diferentes subconjuntos de colunas e então concatena os resultados:

# 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()}")

O parâmetro verbose_feature_names_out (aprimorado na versão 1.6 para aceitar strings e callables) controla a nomeação das features de saída. Defini-lo como True prefixa cada feature com o nome do transformer, evitando colisões de nomes quando múltiplos transformers geram nomes de colunas similares.

Automatizar a Seleção de Colunas com make_column_selector

Codificar nomes de colunas de forma fixa quebra o código quando os datasets mudam. make_column_selector seleciona colunas automaticamente por 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)

O parâmetro remainder controla o que acontece com as colunas não selecionadas por nenhum transformer. As opções incluem 'drop' (padrão), 'passthrough' (manter sem alteração), ou um transformer para aplicar às colunas restantes.

Feature Engineering Dentro dos Pipelines

Os pipelines se estendem além do pré-processamento para incluir etapas de feature engineering. Transformers personalizados herdam de 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())
])

Implementar get_feature_names_out permite que a nova representação HTML do scikit-learn 1.9 exiba os nomes das features de saída, facilitando a depuração e documentação.

Otimização de Hiperparâmetros Através das Etapas do Pipeline

GridSearchCV ajusta parâmetros através de todas as etapas do pipeline usando a convenção de nomenclatura 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}")

O duplo underscore (__) acessa parâmetros aninhados. Para ColumnTransformer, o caminho inclui o nome do transformer: preprocessor__num__imputer__strategy alcança o parâmetro strategy do imputer dentro do transformer numérico.

Cache de Transformers com o Parâmetro memory

Pré-processamento complexo em datasets grandes pode ser lento. O parâmetro memory armazena em cache os transformers ajustados no 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

O cache é particularmente útil durante a busca de hiperparâmetros, onde o mesmo pré-processamento se aplica a múltiplas configurações de modelos. O estimador final (classificador ou regressor) nunca é armazenado em cache, apenas os transformers intermediários.

Monitoramento de Treinamento com Callbacks (scikit-learn 1.9)

A API de callbacks na versão 1.9 adiciona monitoramento de progresso aos pipelines:

# 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_}")

Os callbacks funcionam com Pipeline, StandardScaler, LogisticRegression (solver LBFGS) e todas as classes de busca CV. Esta funcionalidade experimental fornece visibilidade em ajustes de longa duração sem código de logging personalizado.

FeatureUnion para Extração Paralela de Features

FeatureUnion executa múltiplos transformers em paralelo e concatena suas saídas horizontalmente:

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

A versão 1.7.2 adicionou validação garantindo que todos os transformers retornem saídas 2D. Isso detecta erros cedo quando um transformer personalizado acidentalmente retorna um array 1D.

Serialização e Deploy

Os pipelines são serializados com joblib, preservando todos os parâmetros ajustados:

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

O arquivo serializado contém tudo: lógica de seleção de colunas, médias e variâncias ajustadas do scaler, categorias do encoder e pesos do modelo. O deploy requer apenas scikit-learn e joblib, sem código de pré-processamento personalizado.

Perguntas de Entrevista sobre Pipelines do Scikit-Learn

P: O que é vazamento de dados e como os pipelines previnem isso?

O vazamento de dados ocorre quando informações do conjunto de validação ou teste influenciam o treinamento do modelo. O exemplo clássico: ajustar um StandardScaler em todo o dataset antes de dividi-lo. A média e desvio padrão do scaler incluem amostras de teste, tornando os scores de validação cruzada artificialmente altos.

Os pipelines previnem isso encapsulando o pré-processamento dentro do loop de validação cruzada. Quando cross_val_score chama pipeline.fit(), cada fold ajusta o scaler apenas com dados de treino. As estatísticas do fold de teste nunca vazam para o pré-processamento.

P: Como acessar e modificar os parâmetros das etapas aninhadas do pipeline?

Utiliza-se a sintaxe do duplo underscore: pipeline.set_params(classifier__n_estimators=200). Para ColumnTransformer, inclui-se o nome do transformer: pipeline.set_params(preprocessor__num__scaler__with_mean=False). O método get_params() retorna todos os parâmetros aninhados como um dicionário plano, útil para inspeção e logging.

P: Quando usar FeatureUnion vs ColumnTransformer?

ColumnTransformer aplica diferentes transformers a diferentes colunas da mesma entrada. FeatureUnion aplica diferentes transformers a toda a entrada e concatena horizontalmente. Deve-se usar ColumnTransformer para pré-processamento de colunas heterogêneas (numéricas vs categóricas). FeatureUnion é preferível para aumentação de features, onde múltiplos métodos de extração (PCA, features polinomiais, extratores específicos do domínio) produzem features complementares dos mesmos dados.

P: Como depurar um pipeline que falha?

Definir verbose=True no pipeline permite ver os tempos de execução das etapas. É possível acessar resultados intermediários com pipeline.named_steps['step_name'].transform(X) para um pipeline ajustado. No scikit-learn 1.9, a representação HTML mostra atributos ajustados e nomes de features de saída, simplificando a depuração em notebooks Jupyter. Para inspeção mais profunda, usar pipeline[:-1].fit_transform(X, y) para obter a saída logo antes do estimador final.

Checklist de Produção para Pipelines de ML

• Fixar a versão do scikit-learn no requirements para evitar problemas de serialização
• Usar make_column_selector ao invés de nomes de colunas codificados quando os datasets podem mudar
• Definir handle_unknown='ignore' no OneHotEncoder para lidar com novas categorias em produção
• Armazenar em cache o pré-processamento custoso com o parâmetro memory durante o desenvolvimento
• Validar a forma de saída do pipeline e nomes de features nos testes antes do deploy
• Armazenar get_feature_names_out() junto ao modelo para análise de importância de features
• Usar Pipeline dentro de GridSearchCV, nunca o inverso, para garantir comportamento CV correto