Big Data Python

Cuándo usar cada uno

• Pandas: datasets < 1 GB, prototyping, integración con ML/viz
• Polars: datasets 1-100 GB, pipelines de producción, performance crítica
• Dask: datasets > RAM, computación distribuida, clusters

1. Pandas — Referencia rápida

import pandas as pd

df = pd.read_csv("ventas.csv")
df = pd.read_parquet("ventas.parquet")

df.head()
df.info()
df.describe()

# Filtrar
activos = df[df["status"] == "active"]

# Groupby
resumen = df.groupby("region").agg(
    total=("monto", "sum"),
    promedio=("monto", "mean"),
    conteo=("monto", "count"),
).reset_index()

# Join
merged = pd.merge(ventas, clientes, on="client_id", how="left")

# Nuevas columnas
df["margen"] = df["ingreso"] - df["costo"]
df["year"] = pd.to_datetime(df["fecha"]).dt.year

# Guardar
df.to_parquet("resultado.parquet", index=False)
df.to_csv("resultado.csv", index=False)

Operaciones de texto y fechas

df["nombre_upper"] = df["nombre"].str.upper()
df["fecha"] = pd.to_datetime(df["fecha_str"], format="%Y-%m-%d")
df["mes"] = df["fecha"].dt.month

2. Polars — La alternativa rápida

Eager API (similar a Pandas)

import polars as pl

df = pl.read_csv("ventas.csv")
df = pl.read_parquet("ventas.parquet")

df.head()
df.schema
df.describe()

# Filtrar
activos = df.filter(pl.col("status") == "active")

# Groupby
resumen = df.group_by("region").agg(
    pl.col("monto").sum().alias("total"),
    pl.col("monto").mean().alias("promedio"),
    pl.col("monto").count().alias("conteo"),
)

# Join
merged = ventas.join(clientes, on="client_id", how="left")

# Nuevas columnas
df = df.with_columns(
    (pl.col("ingreso") - pl.col("costo")).alias("margen"),
    pl.col("fecha").str.to_date("%Y-%m-%d").dt.year().alias("year"),
)

# Guardar
df.write_parquet("resultado.parquet")
df.write_csv("resultado.csv")

Lazy API (optimización automática)

result = (
    pl.scan_parquet("ventas.parquet")       # no lee aún
    .filter(pl.col("monto") > 1000)         # predicate pushdown
    .group_by("region")
    .agg(pl.col("monto").sum().alias("total"))
    .sort("total", descending=True)
    .collect()                               # ejecuta el plan optimizado
)

La lazy API aplica automáticamente:

• Predicate pushdown: filtra antes de leer todo el archivo
• Projection pushdown: solo lee las columnas necesarias
• Paralelización: distribuye entre todos los cores de CPU

Streaming (datasets más grandes que RAM)

result = (
    pl.scan_parquet("huge_dataset.parquet")
    .filter(pl.col("year") == 2026)
    .group_by("category")
    .agg(pl.col("revenue").sum())
    .collect(engine="streaming")  # procesa en batches
)

3. Dask — Computación distribuida

import dask.dataframe as dd
from dask.distributed import Client

client = Client()  # cluster local automático

df = dd.read_parquet("s3://bucket/ventas/*.parquet")

# API idéntica a Pandas
result = (
    df[df["monto"] > 1000]
    .groupby("region")
    .agg({"monto": "sum"})
    .compute()  # materializa el resultado (devuelve Pandas DF)
)

Con cluster distribuido

from dask.distributed import Client

client = Client("scheduler-address:8786")

df = dd.read_parquet("s3://bucket/data/")
result = df.groupby("category").mean().compute()

4. Conversión entre librerías

# Pandas → Polars
polars_df = pl.from_pandas(pandas_df)

# Polars → Pandas
pandas_df = polars_df.to_pandas()

# Pandas → Dask
dask_df = dd.from_pandas(pandas_df, npartitions=4)

# Dask → Pandas
pandas_df = dask_df.compute()

# Polars ↔ Arrow
arrow_table = polars_df.to_arrow()
polars_df = pl.from_arrow(arrow_table)

5. Ejemplo completo — Pipeline ETL

Con Polars (recomendado para producción)

import polars as pl

pipeline = (
    pl.scan_parquet("raw/ventas_2026_*.parquet")
    .filter(pl.col("status") != "cancelled")
    .with_columns(
        (pl.col("price") * pl.col("quantity")).alias("total"),
        pl.col("date").str.to_date("%Y-%m-%d").alias("fecha"),
    )
    .with_columns(
        pl.col("fecha").dt.month().alias("mes"),
        pl.col("fecha").dt.year().alias("año"),
    )
    .group_by(["región", "mes"])
    .agg(
        pl.col("total").sum().alias("ingresos"),
        pl.col("total").mean().alias("ticket_promedio"),
        pl.len().alias("transacciones"),
    )
    .sort(["región", "mes"])
    .collect()
)

pipeline.write_parquet("processed/resumen_mensual.parquet")
print(pipeline)

Equivalente en Pandas

import pandas as pd
import glob

files = glob.glob("raw/ventas_2026_*.parquet")
df = pd.concat([pd.read_parquet(f) for f in files])

df = df[df["status"] != "cancelled"]
df["total"] = df["price"] * df["quantity"]
df["fecha"] = pd.to_datetime(df["date"])
df["mes"] = df["fecha"].dt.month
df["año"] = df["fecha"].dt.year

resumen = (
    df.groupby(["región", "mes"])
    .agg(
        ingresos=("total", "sum"),
        ticket_promedio=("total", "mean"),
        transacciones=("total", "count"),
    )
    .reset_index()
    .sort_values(["región", "mes"])
)

resumen.to_parquet("processed/resumen_mensual.parquet", index=False)
print(resumen)

Formatos de archivo recomendados

Notas

• Polars es la opción recomendada para nuevos proyectos con datasets medianos-grandes.
• Pandas sigue siendo el estándar cuando se necesita integración con scikit-learn, matplotlib, seaborn.
• Parquet es siempre preferible sobre CSV para datos tabulares.
• Dask es necesario solo para datasets que no caben en RAM o para clusters.
• Docs: https://pandas.pydata.org/docs/ | https://docs.pola.rs/ | https://docs.dask.org/