Saltar a contenido

Pipelines y preprocesado automatizado

1 El problema del preprocesado manual

Cuando aplicas múltiples transformaciones (escalado, imputación, codificación, selección de variables) de forma manual, es fácil cometer errores:

  • Fuga de datos (data leakage): calcular la media con todos los datos (incluido test) antes de separar train/test.
  • Código desordenado: 20 líneas de fit_transform y transform repetidas por todas partes.
  • Difícil de reproducir: si cambias el orden de las transformaciones, cambian los resultados.
  • Imposible en cross-validation: cada fold debería calcular sus propias estadísticas, no las del dataset completo.

Los Pipelines de scikit-learn resuelven todos estos problemas.

2 Pipeline básico

Un Pipeline encadena transformaciones y un modelo final en un único objeto que se comporta como un estimador cualquiera:

from sklearn.pipeline import Pipeline
from sklearn.preprocessing import StandardScaler
from sklearn.decomposition import PCA
from sklearn.ensemble import RandomForestClassifier

pipe = Pipeline([
    ('scaler', StandardScaler()),          # Paso 1: escalar
    ('pca', PCA(n_components=0.95)),        # Paso 2: reducir dimensionalidad
    ('classifier', RandomForestClassifier()) # Paso 3: clasificar
])

# Se usa como un solo modelo
pipe.fit(X_train, y_train)
score = pipe.score(X_test, y_test)

Tip

Ventaja clave: En cross-validation, cada fold aplica el fit solo en los datos de entrenamiento de ese fold, evitando fugas de información. Sin pipelines, esto requeriría código manual muy propenso a errores.

2.1 Acceder a pasos individuales

# Acceder al scaler ajustado
scaler = pipe.named_steps['scaler']
print("Media:", scaler.mean_)
print("Std:", scaler.scale_)

# Acceder al PCA ajustado
pca = pipe.named_steps['pca']
print("Varianza explicada:", pca.explained_variance_ratio_.sum())

# Predicciones
y_pred = pipe.predict(X_test)

2.2 Búsqueda de hiperparámetros con GridSearch

Una de las grandes ventajas de los pipelines es que puedes buscar hiperparámetros de todas las etapas en un solo GridSearchCV:

from sklearn.model_selection import GridSearchCV

param_grid = {
    'pca__n_components': [10, 20, 30, 0.95],     # hiperparámetro de PCA
    'classifier__n_estimators': [100, 200],        # hiperparámetro del clasificador
    'classifier__max_depth': [5, 10, None]         # otro del clasificador
}

grid = GridSearchCV(pipe, param_grid, cv=5, scoring='accuracy')
grid.fit(X_train, y_train)

print(f"Mejores parámetros: {grid.best_params_}")
print(f"Mejor accuracy: {grid.best_score_:.4f}")

3 ColumnTransformer: transformaciones diferentes por tipo de columna

Los datasets reales mezclan columnas numéricas y categóricas que necesitan transformaciones distintas. ColumnTransformer aplica transformaciones diferentes a grupos de columnas:

from sklearn.compose import ColumnTransformer
from sklearn.preprocessing import StandardScaler, OneHotEncoder
from sklearn.impute import SimpleImputer

# Definir qué columnas son numéricas y cuáles categóricas
num_features = ['edad', 'ingresos', 'antiguedad']
cat_features = ['ciudad', 'genero', 'educacion']

preprocessor = ColumnTransformer([
    ('num', Pipeline([
        ('imputer', SimpleImputer(strategy='median')),
        ('scaler', StandardScaler())
    ]), num_features),
    ('cat', Pipeline([
        ('imputer', SimpleImputer(strategy='most_frequent')),
        ('onehot', OneHotEncoder(handle_unknown='ignore'))
    ]), cat_features)
])

# Pipeline completo
pipe = Pipeline([
    ('preprocess', preprocessor),
    ('classifier', RandomForestClassifier(n_estimators=200))
])

pipe.fit(X_train, y_train)
score = pipe.score(X_test, y_test)

3.1 Pipeline completo con ColumnTransformer

from sklearn.compose import ColumnTransformer
from sklearn.pipeline import Pipeline
from sklearn.preprocessing import StandardScaler, OneHotEncoder
from sklearn.impute import SimpleImputer
from sklearn.ensemble import GradientBoostingClassifier

num_features = ['edad', 'ingresos']
cat_features = ['ciudad', 'genero']

# Transformador para numéricas
num_transformer = Pipeline([
    ('imputer', SimpleImputer(strategy='median')),
    ('scaler', StandardScaler())
])

# Transformador para categóricas
cat_transformer = Pipeline([
    ('imputer', SimpleImputer(strategy='most_frequent')),
    ('onehot', OneHotEncoder(handle_unknown='ignore'))
])

# Combinar
preprocessor = ColumnTransformer([
    ('num', num_transformer, num_features),
    ('cat', cat_transformer, cat_features)
])

# Pipeline completo: preprocesado + modelo
full_pipe = Pipeline([
    ('preprocessor', preprocessor),
    ('classifier', GradientBoostingClassifier())
])

full_pipe.fit(X_train, y_train)
print(f"Accuracy: {full_pipe.score(X_test, y_test):.4f}")

4 FeatureUnion: combinar features

FeatureUnion aplica múltiples transformadores en paralelo y concatena sus resultados:

from sklearn.pipeline import FeatureUnion
from sklearn.decomposition import PCA
from sklearn.feature_selection import SelectKBest

# Combinar PCA con selección de características
combined = FeatureUnion([
    ('pca', PCA(n_components=10)),
    ('select', SelectKBest(k=5))
])

X_combined = combined.fit_transform(X_train, y_train)
print(f"Dimensiones originales: {X_train.shape[1]}")
print(f"Dimensiones combinadas: {X_combined.shape[1]}")

5 Pipelines en producción

Nota

¿Por qué los pipelines son esenciales en producción? Porque garantizan que exactamente el mismo preprocesado aplicado al entrenamiento se aplique a los datos nuevos en inferencia. Si escalas con la media del train, necesitas guardar esa media y reusarla. Los pipelines lo hacen automáticamente.

5.1 Guardar y cargar un pipeline

import joblib

# Guardar el pipeline entrenado
joblib.dump(full_pipe, 'pipeline_modelo.pkl')

# Cargar y usar en producción
pipe_cargado = joblib.load('pipeline_modelo.pkl')
predicciones = pipe_cargado.predict(X_nuevos)

5.2 Integración con Flask/FastAPI

# En una API de inferencia
from fastapi import FastAPI
import joblib

app = FastAPI()
pipe = joblib.load('pipeline_modelo.pkl')

@app.post("/predict")
def predict(datos: dict):
    X = pd.DataFrame([datos])
    prediccion = pipe.predict(X)[0]
    return {"clase": prediccion}

6 Errores comunes

Error Causa Solución
Data leakage fit_transform con todos los datos Usar Pipeline con train/test split
Columnas inconsistentes ColumnTransformer no maneja columnas faltantes handle_unknown='ignore'
Memoria agotada Pipeline demasiado grande Usar memory= para cachear transformaciones
Transformadores custom Necesitas lógica propia Crear clase con fit() y transform()

6.1 Transformador custom

Para crear un transformador compatible con scikit-learn, hereda de BaseEstimator (para get_params, set_params) y TransformerMixin (para fit_transform automático):

from sklearn.base import BaseEstimator, TransformerMixin

class DeleteNanRows(BaseEstimator, TransformerMixin):
    def fit(self, X, y=None):
        return self

    def transform(self, X):
        return X.dropna()

# Uso
delete_nan = DeleteNanRows()
X_train_prep = delete_nan.fit_transform(X_train)

6.1.1 Escalar solo columnas seleccionadas

from sklearn.preprocessing import RobustScaler
import pandas as pd

class CustomScaler(BaseEstimator, TransformerMixin):
    def __init__(self, attributes):
        self.attributes = attributes

    def fit(self, X, y=None):
        return self

    def transform(self, X):
        X_copy = X.copy()
        scaler = RobustScaler()
        X_scaled = scaler.fit_transform(X_copy[self.attributes])
        X_scaled = pd.DataFrame(X_scaled, columns=self.attributes, index=X_copy.index)
        for attr in self.attributes:
            X_copy[attr] = X_scaled[attr]
        return X_copy

# Uso: escalar solo columnas específicas
scaler = CustomScaler(['ingresos', 'edad'])
X_train_prep = scaler.fit_transform(X_train)

6.1.2 Codificar solo columnas categóricas

from sklearn.preprocessing import OneHotEncoder

class CustomOneHotEncoding(BaseEstimator, TransformerMixin):
    def __init__(self):
        self._oh = OneHotEncoder(sparse_output=False, handle_unknown='ignore')
        self._columns = None

    def fit(self, X, y=None):
        X_cat = X.select_dtypes(include=['object'])
        self._columns = pd.get_dummies(X_cat).columns
        self._oh.fit(X_cat)
        return self

    def transform(self, X):
        X_copy = X.copy()
        X_cat = X_copy.select_dtypes(include=['object'])
        X_num = X_copy.select_dtypes(exclude=['object'])
        X_cat_oh = self._oh.transform(X_cat)
        X_cat_oh = pd.DataFrame(X_cat_oh, columns=self._columns, index=X_copy.index)
        X_copy.drop(list(X_cat), axis=1, inplace=True)
        return X_copy.join(X_cat_oh)

# Uso
encoder = CustomOneHotEncoding()
X_train_encoded = encoder.fit_transform(X_train)

6.2 FunctionTransformer: transformaciones simples

Para transformaciones que no necesitan fit() (son puramente matemáticas), FunctionTransformer envuelve cualquier función en un estimador compatible con pipelines:

from sklearn.preprocessing import FunctionTransformer
import numpy as np

# Transformación de valores absolutos
abs_transformer = FunctionTransformer(np.abs, validate=False)
abs_transformer.fit_transform(np.array([[-1, 2], [-3, 4]]))
# [[1, 2], [3, 4]]

# Función personalizada: elevar al cuadrado
def cuadrado(x):
    return x ** 2

cuadrado_transformer = FunctionTransformer(cuadrado, validate=False)
cuadrado_transformer.fit_transform(np.array([[1, 2], [3, 4]]))
# [[1, 4], [9, 16]]

# Logaritmo (para datos con cola pesada)
log_transformer = FunctionTransformer(np.log1p, validate=False)

# Integrado en un pipeline
from sklearn.pipeline import Pipeline
from sklearn.preprocessing import StandardScaler
from sklearn.ensemble import RandomForestClassifier

pipe = Pipeline([
    ('log', FunctionTransformer(np.log1p, validate=False)),
    ('scaler', StandardScaler()),
    ('model', RandomForestClassifier())
])

Tip

¿Cuándo usar FunctionTransformer en vez de BaseEstimator? Cuando la transformación es pura función (log, abs, cuadrado) y no necesitas aprender nada de los datos. Si necesitas fit() para calcular algo (media, frecuencias, etc.), usa BaseEstimator + TransformerMixin.

7 Resumen

Componente Función Ejemplo
Pipeline Encadenar pasos secuencialmente Escalar → PCA → Clasificar
ColumnTransformer Transformaciones diferentes por columna Num → StandardScaler, Cat → OneHot
FeatureUnion Combinar features en paralelo PCA + SelectKBest
Transformador custom Lógica de preprocesado propia CustomScaler, CustomOneHotEncoding
FunctionTransformer Envolver funciones simples log, abs, cuadrado

Tip

Regla de oro: Un pipeline expone los métodos del último estimador. Si el último es un clasificador → fit, predict. Si es un transformador → fit, transform, fit_transform.

Los pipelines son obligatorios en cualquier proyecto serio de machine learning. Evitan fugas de datos, simplifican el código y hacen que el modelo sea fácil de desplegar en producción.