Ingeniería de Datos con Python: Polars y DuckDB

Ingeniería de Datos con Python: Polars y DuckDB

Pandas es la herramienta estándar para análisis de datos en Python, pero cuando los datasets crecen a cientos de millones de filas o necesitas consultas SQL complejas con latencia mínima, Polars y DuckDB son las alternativas modernas que debes conocer. Ambas están escritas en Rust/C++ y aprovechan el procesamiento columnar para ser 10-100× más rápidas que pandas en operaciones de agregación y filtrado.

Por qué Polars y DuckDB

pip install polars duckdb pyarrow

Polars: DataFrames rápidos con evaluación lazy

Operaciones básicas eager

import polars as pl

# Leer CSV (mucho más rápido que pandas)
df = pl.read_csv("ventas.csv")

# Selección y filtrado
resultado = df.filter(
    (pl.col("importe") > 1000) & (pl.col("pais") == "ES")
).select([
    "fecha", "cliente", "importe", "producto"
]).sort("importe", descending=True)

print(resultado.head(10))
print(resultado.describe())

Evaluación lazy: el superpoder de Polars

# LazyFrame — no ejecuta nada hasta .collect()
lf = pl.scan_csv("ventas_100M.csv")  # archivo de 100M filas

pipeline = (
    lf
    .filter(pl.col("año") >= 2023)
    .group_by(["pais", "producto"])
    .agg([
        pl.col("importe").sum().alias("total"),
        pl.col("importe").mean().alias("media"),
        pl.col("cliente").n_unique().alias("clientes_unicos"),
        pl.col("importe").count().alias("transacciones"),
    ])
    .sort("total", descending=True)
    .limit(50)
)

# Ver el plan de ejecución optimizado
print(pipeline.explain())

# Ejecutar (aquí sí se procesa)
resultado = pipeline.collect()
print(resultado)

Expresiones y funciones de ventana

df = pl.read_parquet("datos.parquet")

# Columnas calculadas con expresiones
df = df.with_columns([
    # Ratio sobre media global
    (pl.col("ventas") / pl.col("ventas").mean()).alias("ratio_media"),

    # Rank por grupo (ventana)
    pl.col("ventas")
      .rank(method="dense", descending=True)
      .over("region")
      .alias("rank_region"),

    # Media móvil de 7 días
    pl.col("ventas")
      .rolling_mean(window_size=7, min_periods=1)
      .alias("media_7d"),

    # Diferencia con el día anterior
    pl.col("ventas").diff(1).alias("delta_ventas"),

    # Clasificación condicional
    pl.when(pl.col("ventas") > 10000)
      .then(pl.lit("alto"))
      .when(pl.col("ventas") > 5000)
      .then(pl.lit("medio"))
      .otherwise(pl.lit("bajo"))
      .alias("segmento"),
])

Uniones y operaciones de join

clientes = pl.read_csv("clientes.csv")
pedidos  = pl.read_csv("pedidos.csv")
productos = pl.read_csv("productos.csv")

# Join encadenado
resultado = (
    pedidos
    .join(clientes, on="cliente_id", how="left")
    .join(productos, on="producto_id", how="inner")
    .filter(pl.col("categoria") == "electronica")
    .group_by("pais")
    .agg(pl.col("importe").sum().alias("total_electronica"))
    .sort("total_electronica", descending=True)
)

Tipos de datos y casting

df = df.with_columns([
    pl.col("fecha").str.to_datetime("%Y-%m-%d"),
    pl.col("precio").cast(pl.Float64),
    pl.col("cantidad").cast(pl.UInt32),
    pl.col("codigo").cast(pl.Categorical),
])

# Extraer partes de fecha
df = df.with_columns([
    pl.col("fecha").dt.year().alias("año"),
    pl.col("fecha").dt.month().alias("mes"),
    pl.col("fecha").dt.weekday().alias("dia_semana"),
])

Formatos de archivo eficientes

# Parquet — formato columnar comprimido (recomendado para datos grandes)
df.write_parquet("datos.parquet", compression="zstd")
df2 = pl.read_parquet("datos.parquet")

# Leer múltiples archivos Parquet (glob)
df_all = pl.read_parquet("datos_particion_*.parquet")

# Arrow IPC (más rápido para transferencia entre procesos)
df.write_ipc("datos.arrow")
df3 = pl.read_ipc("datos.arrow")

# CSV con opciones
df.write_csv("salida.csv", separator=";")

DuckDB: SQL OLAP embebido en Python

DuckDB es una base de datos analítica in-process — no necesita servidor. Funciona directamente sobre DataFrames de pandas/Polars, archivos Parquet, CSV, JSON e incluso datos en memoria.

Instalación y uso básico

import duckdb

# Conexión en memoria (default)
con = duckdb.connect()

# O persistida en disco
con = duckdb.connect("analytics.duckdb")

# Consulta directa sobre archivo
result = con.execute("""
    SELECT
        pais,
        SUM(importe) as total,
        COUNT(*) as pedidos,
        AVG(importe) as ticket_medio
    FROM 'ventas.csv'
    WHERE año >= 2023
    GROUP BY pais
    ORDER BY total DESC
    LIMIT 20
""").df()  # .df() → pandas DataFrame, .pl() → Polars DataFrame

print(result)

DuckDB + Polars (integración nativa)

import duckdb
import polars as pl

# Registrar un DataFrame de Polars como tabla virtual
df_polars = pl.read_parquet("ventas.parquet")

con = duckdb.connect()
# DuckDB puede leer DataFrames de Polars directamente via Arrow
result = con.execute("""
    SELECT
        producto,
        region,
        SUM(ventas) as total_ventas,
        RANK() OVER (PARTITION BY region ORDER BY SUM(ventas) DESC) as rank
    FROM df_polars
    GROUP BY producto, region
    QUALIFY rank <= 3
""").pl()  # devuelve Polars DataFrame

print(result)

Consultas sobre múltiples archivos Parquet

# DuckDB puede leer particiones completas con glob
con = duckdb.connect()

resultado = con.execute("""
    SELECT
        YEAR(fecha) as año,
        MONTH(fecha) as mes,
        SUM(importe) as ingresos,
        COUNT(DISTINCT cliente_id) as clientes_activos
    FROM read_parquet('datos/año=*/mes=*/*.parquet', hive_partitioning=true)
    WHERE año BETWEEN 2022 AND 2024
    GROUP BY 1, 2
    ORDER BY 1, 2
""").df()

Window functions y CTEs avanzados

query = """
WITH ventas_acumuladas AS (
    SELECT
        fecha,
        region,
        importe,
        SUM(importe) OVER (
            PARTITION BY region
            ORDER BY fecha
            ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW
        ) as acumulado,
        LAG(importe, 1, 0) OVER (PARTITION BY region ORDER BY fecha) as dia_anterior,
        AVG(importe) OVER (
            PARTITION BY region
            ORDER BY fecha
            ROWS BETWEEN 6 PRECEDING AND CURRENT ROW
        ) as media_7d
    FROM ventas
),
variacion AS (
    SELECT *,
        ROUND(100.0 * (importe - dia_anterior) / NULLIF(dia_anterior, 0), 2) as variacion_pct
    FROM ventas_acumuladas
)
SELECT * FROM variacion
WHERE fecha >= CURRENT_DATE - INTERVAL 30 DAY
ORDER BY fecha DESC, region
"""

df = con.execute(query).df()

Pipeline ETL completo con DuckDB

import duckdb
import polars as pl
from pathlib import Path
import time

def pipeline_etl(carpeta_entrada: str, archivo_salida: str):
    """Pipeline ETL: CSV crudos → Parquet limpio y agregado."""
    con = duckdb.connect()

    inicio = time.perf_counter()

    # 1. Extraer: leer todos los CSV de la carpeta
    con.execute(f"""
        CREATE TABLE raw AS
        SELECT * FROM read_csv_auto('{carpeta_entrada}/*.csv',
            union_by_name=true,
            ignore_errors=true)
    """)

    n_raw = con.execute("SELECT COUNT(*) FROM raw").fetchone()[0]
    print(f"Filas extraídas: {n_raw:,}")

    # 2. Transformar: limpiar, normalizar y enriquecer
    con.execute("""
        CREATE TABLE clean AS
        SELECT
            CAST(fecha AS DATE)                        AS fecha,
            UPPER(TRIM(pais))                          AS pais,
            COALESCE(region, 'DESCONOCIDO')            AS region,
            CAST(REPLACE(importe, ',', '.') AS DOUBLE) AS importe,
            CAST(cantidad AS INTEGER)                  AS cantidad,
            importe * cantidad                         AS importe_total,
            MD5(CONCAT(fecha, cliente_id, producto_id)) AS id_unico
        FROM raw
        WHERE fecha IS NOT NULL
          AND importe > 0
          AND cantidad > 0
        QUALIFY ROW_NUMBER() OVER (PARTITION BY id_unico ORDER BY fecha) = 1
    """)

    n_clean = con.execute("SELECT COUNT(*) FROM clean").fetchone()[0]
    print(f"Filas limpias: {n_clean:,} ({100*n_clean/n_raw:.1f}%)")

    # 3. Cargar: escribir Parquet particionado
    con.execute(f"""
        COPY (
            SELECT * FROM clean
            ORDER BY pais, fecha
        ) TO '{archivo_salida}'
        (FORMAT PARQUET, COMPRESSION ZSTD, ROW_GROUP_SIZE 100000)
    """)

    elapsed = time.perf_counter() - inicio
    size_mb = Path(archivo_salida).stat().st_size / 1_048_576
    print(f"Parquet escrito: {size_mb:.1f} MB en {elapsed:.2f}s")

    return con.execute("SELECT * FROM clean LIMIT 5").df()

# Ejecutar pipeline
muestra = pipeline_etl("datos/raw", "datos/clean.parquet")
print(muestra)

Comparativa de rendimiento: pandas vs Polars vs DuckDB

import pandas as pd
import polars as pl
import duckdb
import time
import numpy as np

# Generar dataset de prueba (10M filas)
np.random.seed(42)
n = 10_000_000

datos = {
    "id": np.arange(n),
    "pais": np.random.choice(["ES", "MX", "AR", "CO", "PE"], n),
    "importe": np.random.exponential(500, n),
    "categoria": np.random.choice(["A", "B", "C", "D"], n),
}

def benchmark(nombre, fn):
    inicio = time.perf_counter()
    resultado = fn()
    elapsed = time.perf_counter() - inicio
    print(f"{nombre:20s}: {elapsed:.3f}s — {resultado.shape if hasattr(resultado, 'shape') else len(resultado)} filas")
    return elapsed

# pandas
df_pd = pd.DataFrame(datos)
benchmark("pandas", lambda:
    df_pd.groupby(["pais", "categoria"])["importe"]
         .agg(["sum", "mean", "count"])
         .reset_index()
)

# Polars eager
df_pl = pl.DataFrame(datos)
benchmark("polars eager", lambda:
    df_pl.group_by(["pais", "categoria"])
         .agg([pl.col("importe").sum(), pl.col("importe").mean(), pl.col("importe").count()])
)

# DuckDB
benchmark("duckdb", lambda:
    duckdb.execute("""
        SELECT pais, categoria, SUM(importe), AVG(importe), COUNT(*)
        FROM df_pd GROUP BY pais, categoria
    """).df()
)

# Resultados típicos en un portátil moderno:
# pandas          :  2.841s
# polars eager    :  0.187s  (15× más rápido)
# duckdb          :  0.142s  (20× más rápido)

Cuándo usar cada herramienta

Integración con el ecosistema de datos

# Polars → pandas (cuando necesitas una librería solo compatible con pandas)
df_pandas = df_polars.to_pandas()
df_polars = pl.from_pandas(df_pandas)

# Polars → Arrow (para transferencia entre sistemas)
tabla_arrow = df_polars.to_arrow()
df_polars2  = pl.from_arrow(tabla_arrow)

# DuckDB → Parquet directamente
con.execute("COPY tabla TO 'salida.parquet' (FORMAT PARQUET, COMPRESSION ZSTD)")

# Leer desde S3 (con extensión httpfs)
con.execute("INSTALL httpfs; LOAD httpfs;")
df = con.execute("""
    SELECT * FROM read_parquet('s3://mi-bucket/datos/*.parquet')
    LIMIT 1000
""").pl()

Conclusión

Polars y DuckDB son herramientas complementarias que cubren la mayor parte de las necesidades de ingeniería de datos moderna sin necesidad de un clúster Spark. Polars brilla en transformaciones encadenadas con evaluación lazy, mientras que DuckDB sobresale en consultas SQL analíticas ad-hoc. Ambas se integran perfectamente con el ecosistema Arrow/Parquet, hacen uso de todos los núcleos de la CPU y consumen mucha menos memoria que pandas para operaciones de agregación. Para pipelines de datos en Python en 2025, esta combinación es el punto de partida recomendado antes de escalar a Spark o plataformas cloud.

Conversiones relacionadas

Conversiones frecuentes del catálogo:

• JPG a PNG
• MP4 a MP3
• PDF a DOCX
• PNG a JPG
• MP3 a WAV
• DOCX a PDF