Construir un Pipeline ETL con pandas y Parquet

Visión General

pandas es la herramienta estándar para procesamiento de datos tabulares en Python. Parquet es un formato de almacenamiento columnar que comprime mejor que CSV y preserva los tipos de datos (enteros, floats, datetimes, categoricals). Combinarlos en un pipeline ETL te da procesamiento de datos type-safe con almacenamiento compacto. Los ejemplos de abajo muestran cómo extraer de múltiples fuentes, transformar con coerción de tipos y validación, y cargar a archivos Parquet particionados.

Cuándo Usar

• Trabajos de procesamiento de datos por lotes que se ejecutan programados (cada hora, cada día)
• Convertir exportaciones CSV/JSON en Parquet tipado para analítica posterior
• Pipelines de datos donde los archivos intermedios deben preservar los tipos
• Construir variables para modelos de ML a partir de fuentes de datos en bruto
• Cualquier escenario donde necesitas transformaciones de datos reproducibles y auditables

Cuándo No Usar

• Pipelines de streaming/tiempo real — usa Spark Structured Streaming o Flink
• Datasets más grandes que la memoria — usa Polars, Dask o PySpark
• Transformaciones puntuales simples — basta un pd.read_csv().to_parquet()
• Almacenes de datos analíticos — usa dbt para transformaciones basadas en SQL

Solución

Los snippets de abajo están listos para copiar. Una versión ejecutable que genera datos de muestra y corre todo el pipeline de punta a punta está en el repositorio complementario.

Estructura básica de pipeline ETL

import pandas as pd
from pathlib import Path
import logging

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)


def extract_csv(path: str) -> pd.DataFrame:
    """Extraer datos de un archivo CSV."""
    logger.info(f"Extracting from {path}")
    df = pd.read_csv(path)
    logger.info(f"Extracted {len(df)} rows, {len(df.columns)} columns")
    return df


def extract_json(path: str) -> pd.DataFrame:
    """Extraer datos de un archivo JSON."""
    logger.info(f"Extracting from {path}")
    df = pd.read_json(path, lines=True)
    logger.info(f"Extracted {len(df)} rows")
    return df


def transform(df: pd.DataFrame) -> pd.DataFrame:
    """Aplicar transformaciones: coerción de tipos, limpieza, columnas derivadas."""
    logger.info("Starting transformation")

    # Coerción de tipos
    df["order_date"] = pd.to_datetime(df["order_date"], errors="coerce")
    df["amount"] = pd.to_numeric(df["amount"], errors="coerce")
    df["quantity"] = pd.to_numeric(df["quantity"], errors="coerce").astype("Int64")

    # Eliminar filas con fechas o montos inválidos
    df = df.dropna(subset=["order_date", "amount"])

    # Columnas derivadas
    df["year"] = df["order_date"].dt.year
    df["month"] = df["order_date"].dt.month
    df["revenue"] = df["amount"] * df["quantity"]

    # Normalizar columnas de texto
    df["customer_name"] = df["customer_name"].str.strip().str.title()

    # Categorical para columnas de baja cardinalidad
    df["status"] = df["status"].astype("category")

    logger.info(f"Transformed to {len(df)} rows")
    return df


def load_parquet(df: pd.DataFrame, path: str, partition_cols: list[str] | None = None) -> None:
    """Cargar DataFrame a Parquet, opcionalmente particionado."""
    logger.info(f"Loading to {path}")
    if partition_cols:
        df.to_parquet(path, partition_cols=partition_cols, index=False)
    else:
        df.to_parquet(path, index=False)
    logger.info(f"Loaded {len(df)} rows")


def run_pipeline(source_path: str, destination_path: str) -> None:
    """Ejecutar el pipeline ETL completo."""
    df = extract_csv(source_path)
    df = transform(df)
    load_parquet(df, destination_path, partition_cols=["year", "month"])


if __name__ == "__main__":
    run_pipeline("data/raw/orders.csv", "data/processed/orders")

Extraer de múltiples fuentes y combinar

def extract_and_merge(orders_path: str, customers_path: str) -> pd.DataFrame:
    """Extraer de múltiples fuentes y combinar."""
    orders = pd.read_csv(orders_path)
    customers = pd.read_csv(customers_path)

    # Estandarizar claves de unión
    orders["customer_id"] = orders["customer_id"].astype(str).str.strip()
    customers["customer_id"] = customers["customer_id"].astype(str).str.strip()

    merged = orders.merge(customers, on="customer_id", how="left")

    logger.info(f"Merged: {len(orders)} orders + {len(customers)} customers = {len(merged)} rows")
    return merged

Transformar con validación

def transform_with_validation(df: pd.DataFrame) -> pd.DataFrame:
    """Transformar con comprobaciones de calidad de datos."""
    # Coerción de tipos
    df["order_date"] = pd.to_datetime(df["order_date"], errors="coerce")
    df["amount"] = pd.to_numeric(df["amount"], errors="coerce")

    # Validación: sin montos negativos
    negative_count = (df["amount"] < 0).sum()
    if negative_count > 0:
        logger.warning(f"Found {negative_count} negative amounts, filtering out")
        df = df[df["amount"] >= 0]

    # Validación: sin IDs de pedido duplicados
    dup_count = df.duplicated(subset=["order_id"]).sum()
    if dup_count > 0:
        logger.warning(f"Found {dup_count} duplicate order IDs, dropping duplicates")
        df = df.drop_duplicates(subset=["order_id"], keep="last")

    # Validación: columnas requeridas presentes
    required_cols = ["order_id", "customer_id", "order_date", "amount"]
    missing = [c for c in required_cols if c not in df.columns]
    if missing:
        raise ValueError(f"Missing required columns: {missing}")

    # Columnas derivadas
    df["year"] = df["order_date"].dt.year
    df["month"] = df["order_date"].dt.month
    df["quarter"] = df["order_date"].dt.quarter

    return df

Salida Parquet particionada

def load_partitioned(df: pd.DataFrame, base_path: str) -> None:
    """Cargar a Parquet particionado por año y mes."""
    # Asegurar que las columnas de partición sean strings (requerimiento de Parquet)
    df["year"] = df["year"].astype(str)
    df["month"] = df["month"].astype(str).str.zfill(2)

    df.to_parquet(
        base_path,
        partition_cols=["year", "month"],
        index=False,
        engine="pyarrow",
        compression="snappy",
    )
    logger.info(f"Partitioned output at {base_path}/year=*/month=*")


def read_partitioned(base_path: str, year: str, month: str | None = None) -> pd.DataFrame:
    """Leer particiones específicas."""
    if month:
        path = f"{base_path}/year={year}/month={month}"
    else:
        path = f"{base_path}/year={year}"
    return pd.read_parquet(path)

Carga incremental (anexar a Parquet existente)

def load_incremental(df: pd.DataFrame, path: str) -> None:
    """Anexar datos nuevos a un dataset Parquet existente."""
    from pathlib import Path

    if Path(path).exists():
        existing = pd.read_parquet(path)
        combined = pd.concat([existing, df], ignore_index=True)
        combined = combined.drop_duplicates(subset=["order_id"], keep="last")
    else:
        combined = df

    combined.to_parquet(path, index=False)
    logger.info(f"Incremental load: {len(df)} new rows, {len(combined)} total")

Pipeline con manejo de errores y reintentos

import time

def extract_with_retry(path: str, retries: int = 3, delay: int = 5) -> pd.DataFrame:
    """Extraer con lógica de reintento para fuentes de red."""
    for attempt in range(retries):
        try:
            if path.startswith("http"):
                df = pd.read_csv(path)
            else:
                df = pd.read_csv(path)
            return df
        except Exception as e:
            logger.warning(f"Attempt {attempt + 1}/{retries} failed: {e}")
            if attempt < retries - 1:
                time.sleep(delay * (attempt + 1))
            raise


def run_pipeline_safe(source: str, destination: str) -> bool:
    """Ejecutar el pipeline con manejo de errores completo."""
    try:
        df = extract_with_retry(source)
        df = transform_with_validation(df)
        load_partitioned(df, destination)
        logger.info("Pipeline completed successfully")
        return True
    except Exception as e:
        logger.error(f"Pipeline failed: {e}")
        return False

Validación de esquema

EXPECTED_SCHEMA = {
    "order_id": "int64",
    "customer_id": "object",
    "order_date": "datetime64[ns]",
    "amount": "float64",
    "quantity": "Int64",
    "status": "category",
}

def enforce_schema(df: pd.DataFrame) -> pd.DataFrame:
    """Aplicar el esquema esperado en el DataFrame."""
    for col, dtype in EXPECTED_SCHEMA.items():
        if col not in df.columns:
            raise ValueError(f"Missing column: {col}")
        if df[col].dtype != dtype:
            logger.info(f"Converting {col} from {df[col].dtype} to {dtype}")
            if dtype == "datetime64[ns]":
                df[col] = pd.to_datetime(df[col], errors="coerce")
            elif dtype == "category":
                df[col] = df[col].astype("category")
            else:
                df[col] = df[col].astype(dtype)
    return df

Explicación

El pipeline es un grafo lineal con una rama de fallo: cada etapa consume la salida de la anterior y registra conteos de filas, así puedes ver exactamente dónde se perdieron registros.

Por qué Int64 y no int64: errors="coerce" convierte los valores imposibles de parsear en NaN, y el dtype plano int64 no puede contener NaN. El propio cast fallaría. El dtype nullable Int64 almacena pd.NA, así quantity sobrevive a las filas cuyo valor original no era numérico.

Por qué particionar por año/mes: partition_cols escribe una estructura de directorios (year=2026/month=10/) que los motores de consulta usan para podar particiones — una consulta filtrada a un mes lee solo los archivos de esa carpeta en vez de escanear todo el dataset. Mantén la granularidad al nivel por el que realmente filtran tus consultas: particiones diarias sobre datos pequeños generan miles de archivos diminutos, que terminan siendo más lentos que un archivo mediano.

Qué códec de compresión elegir:

Códec	Velocidad	Ratio	Úsalo cuando
snappy	La más rápida	El más bajo	Los archivos se leen a menudo — es el default por algo
zstd	Rápido	~2x mejor que snappy	Datasets de archivo o limitados por almacenamiento
gzip	La más lenta	Bueno	Una herramienta downstream no puede leer snappy/zstd

Los reintentos van en la E/S, no en las transformaciones: una transformación que falla por datos malos fallará en cada reintento, mientras que un sistema de archivos o fuente de red inestable suele funcionar al segundo intento. Por eso el wrapper de reintentos envuelve solo extract: los errores de transformación salen a la superficie de inmediato en vez de quemar reintentos.

Para reglas de esquema declarativas y comprobaciones por columna más allá de lo que hace transform_with_validation aquí, consulta Validar schemas de DataFrame con Pandera.

Variantes

Usar PyArrow directamente para archivos grandes

import pyarrow.parquet as pq
import pyarrow as pa

def load_with_pyarrow(df: pd.DataFrame, path: str) -> None:
    """Escribir Parquet usando PyArrow para más control."""
    table = pa.Table.from_pandas(df, preserve_index=False)
    pq.write_table(
        table,
        path,
        compression="zstd",
        compression_level=3,
        use_dictionary=True,
        write_statistics=True,
    )

Pipeline con registro a archivo

import logging.handlers

def setup_logging(log_path: str = "logs/etl.log") -> None:
    """Configurar registro a archivo + consola."""
    handler = logging.handlers.RotatingFileHandler(
        log_path, maxBytes=10_000_000, backupCount=5
    )
    handler.setFormatter(logging.Formatter(
        "%(asctime)s - %(name)s - %(levelname)s - %(message)s"
    ))
    logging.basicConfig(
        level=logging.INFO,
        handlers=[handler, logging.StreamHandler()],
    )

Orquestación del pipeline con configuración

import yaml

def load_config(config_path: str) -> dict:
    with open(config_path) as f:
        return yaml.safe_load(f)

def run_pipeline_from_config(config_path: str) -> None:
    config = load_config(config_path)

    df = pd.read_csv(config["source"])
    for transform_config in config.get("transforms", []):
        if transform_config["type"] == "rename":
            df = df.rename(columns=transform_config["mapping"])
        elif transform_config["type"] == "filter":
            df = df.query(transform_config["condition"])
        elif transform_config["type"] == "cast":
            df[transform_config["column"]] = df[transform_config["column"]].astype(
                transform_config["dtype"]
            )

    df.to_parquet(
        config["destination"],
        partition_cols=config.get("partition_cols"),
        index=False,
    )

Mejores Prácticas

• Para una guía más detallada, consulta Programar y Monitorear DAGs con Apache Airflow.
• Usa errors="coerce" en pd.to_numeric y pd.to_datetime: un valor malformado se convierte en NaN, que puedes contar y eliminar, en vez de una excepción que mata toda la ejecución
• Particiona por columnas de fecha (año, mes) — los lectores pueden podar hasta rangos de tiempo específicos en vez de escanear todos los archivos
• Usa compresión snappy para velocidad, zstd para mejor ratio de compresión — elige según la tabla de códecs de arriba
• Registra conteos de filas en cada etapa: el log te dice qué etapa descartó filas sin depuración extra
• Valida los datos antes de escribir — una fila mala detectada en la transformación es barata; esa misma fila en Parquet contamina cada lectura posterior
• Usa Int64 (entero nullable) en lugar de int64 cuando los datos pueden tener valores faltantes — int64 no puede representar NaN y el cast falla a mitad del pipeline
• Escribe estadísticas en Parquet — los metadatos min/max permiten predicate pushdown y que los lectores salten grupos de filas que no pueden coincidir

Errores Comunes

• No manejar valores faltantes: pd.to_numeric sin errors="coerce" lanza error en el primer string inválido que encuentra — a menudo a medio camino de un archivo grande. Coacciona primero, después decide si eliminar o rellenar los NaN resultantes.
• Usar CSV como formato intermedio: CSV guarda todo como texto, así que tus dtypes cuidadosos desaparecen en la siguiente lectura. Parquet los preserva.
• No particionar datasets grandes: un solo archivo Parquet de 10GB obliga a cada consulta a leerlo todo. Particiona por la columna por la que más filtran tus consultas.
• Ignorar los dtypes después de leer: pd.read_csv infiere tipos y a menudo adivina mal — los IDs se vuelven floats, los códigos postales pierden ceros iniciales. Haz cast explícito de las columnas que te importan.
• No deduplicar en cargas incrementales: re-ejecutar un job fallido anexa el mismo lote dos veces. Deduplica por una clave natural como order_id antes de escribir.

Lecturas Adicionales

• Referencia de to_parquet de pandas — opciones de partition_cols, compresión y motor
• Módulo parquet de PyArrow — grupos de filas, estadísticas y control de escritura de bajo nivel
• Documentación del formato Apache Parquet — codificaciones, compresión y layout de archivos
• Herramientas de E/S de pandas: particionar datasets — cómo pandas maneja los layouts de directorios clave=valor