Operaciones Paralelas de DataFrame con Dask

Visión General

Dask se apoya en pandas y NumPy para trabajar con datasets más grandes que la memoria. Corta un DataFrame en particiones, y cada partición es un pandas DataFrame normal. Dask construuye un task graph de operaciones y las ejecuta de forma lazy, paralelizando entre cores con el scheduler local o entre máquinas con el scheduler distribuido. Como la API refleja la de pandas, la mayoría del código existente funciona con cambios mínimos. Si venís de un pipeline ETL con pandas, la migración es mayormente cambiar pd por dd y agregar .compute().

Cuándo Usar

Dask rinde cuando los datasets van de unos pocos gigabytes a aproximadamente un terabyte, demasiado grandes para la memoria pero manejables en disco. Si ya tenés código de pandas que necesita escalar sin una reescritura grande, la API familiar de Dask es una victoria rápida. También sirve cuando querés paralelismo sin el overhead de un cluster de Spark, o cuando tus pipelines ETL mueven archivos Parquet, CSV o HDF5. Para workloads a escala Spark con group-by pesado, consultá la receta de group-by aggregation en Spark que cubre el lado JVM. Por último, usá Dask cuando necesitás lógica paralela custom que va más allá de group-by y join.

Cuándo Evitar

Evitá Dask cuando los datasets sean menores a 1 GB. pandas suele ser más rápido porque evita el overhead del task graph de Dask. También evitalo si necesitás el ecosistema completo de pandas, porque Dask no implementa todos los métodos de pandas. Para streams o procesamiento en tiempo real, Flink o Structured Streaming son un mejor ajuste. Y si Polars es suficiente, usalo: suele ser más rápido y ligero para la mayoría del trabajo con DataFrames.

Solución

Dask DataFrame Básico

import dask.dataframe as dd

# Leer CSV lazy — no carga hasta compute()
ddf = dd.read_csv("data/orders_*.csv")

# Leer Parquet
ddf = dd.read_parquet("data/orders/")

# Desde un pandas DataFrame
import pandas as pd
pdf = pd.read_csv("data.csv")
ddf = dd.from_pandas(pdf, npartitions=4)

# Inspeccionar particiones
print(ddf.npartitions)  # número de particiones
print(ddf.divisions)    # límites de particiones, conocidos si están ordenados

Operaciones Lazy

# Construir task graph — sin ejecución todavía
result = (
    ddf
    .query("amount > 100")
    .groupby("customer_id")
    .agg({"amount": "sum"})
    .reset_index()
    .sort_values("amount", ascending=False)
)

# Ejecutar el graph y obtener un pandas DataFrame
df = result.compute()
print(df.head(10))

Lectura y Escritura

# Leer múltiples archivos CSV
ddf = dd.read_csv("data/2025-*.csv", parse_dates=["order_date"])

# Leer con dtypes explícitos
ddf = dd.read_csv(
    "data/orders.csv",
    dtype={
        "order_id": "int64",
        "amount": "float64",
        "customer_id": "object",
    },
)

# Escribir Parquet (un archivo por partición por defecto)
ddf.to_parquet("data/output/", write_index=False)

# Escribir CSV con un archivo por partición
ddf.to_csv("data/output_*.csv", index=False)

Group-by y Agregación

# Agregación group-by en paralelo
result = (
    ddf
    .groupby("customer_id")
    .agg({
        "amount": ["sum", "mean", "count"],
        "order_id": "nunique",
    })
    .compute()
)

# Agregaciones con nombres custom
result = (
    ddf
    .groupby("category")
    .agg(
        total_revenue=("amount", "sum"),
        avg_order=("amount", "mean"),
        order_count=("order_id", "count"),
    )
    .compute()
)

Joins

orders = dd.read_parquet("data/orders/")
customers = dd.read_parquet("data/customers/")

# Merge estándar; puede requerir un shuffle si no está ordenado por la join key
joined = orders.merge(customers, on="customer_id", how="left")

# Broadcast join cuando el lado derecho es chico
small_customers = customers.head(1000)  # pandas DataFrame
joined = orders.merge(
    dd.from_pandas(small_customers, npartitions=1),
    on="customer_id",
    how="left",
    broadcast=True,
)

result = joined.compute()

Cómputo Custom con map_partitions

def process_partition(pdf: pd.DataFrame) -> pd.DataFrame:
    pdf["amount_with_tax"] = pdf["amount"] * 1.1
    pdf["order_date"] = pd.to_datetime(pdf["order_date"])
    pdf["month"] = pdf["order_date"].dt.month
    return pdf

ddf_processed = ddf.map_partitions(process_partition)
result = ddf_processed.compute()

Task Graphs Custom con delayed

import dask

@dask.delayed
def load_file(path):
    return pd.read_csv(path)

@dask.delayed
def clean(df):
    df["amount"] = pd.to_numeric(df["amount"], errors="coerce")
    return df.dropna(subset=["amount"])

@dask.delayed
def combine(dfs):
    return pd.concat(dfs).groupby("customer_id")["amount"].sum()

files = ["data/jan.csv", "data/feb.csv", "data/mar.csv"]
processed = [clean(load_file(f)) for f in files]
result = combine(processed)

df = result.compute()

Repartitioning

# Setear número de particiones
ddf = ddf.repartition(npartitions=10)

# O setear tamaño de partición
ddf = ddf.repartition(partition_size="100MB")

# Setear un índice ordenado para que divisions sean conocidas
ddf = ddf.reset_index(drop=True)
ddf = ddf.set_index("customer_id")  # dispara un shuffle

Usar Dask Distributed

from dask.distributed import Client

# Cluster local usando todos los cores
client = Client(n_workers=4, threads_per_worker=2, memory_limit="4GB")

# Todos los .compute() usan el distributed scheduler
ddf = dd.read_parquet("data/orders/")
result = ddf.groupby("customer_id")["amount"].sum().compute()

# Cerrar el client al terminar
client.close()

Persistir Data en Memoria

# Cargar a memoria distribuida entre workers
ddf_persisted = ddf.persist()

# Reusar el objeto persistido para operaciones repetidas más rápidas
result = ddf_persisted.groupby("customer_id")["amount"].sum().compute()

Monitoreo de Progreso

from dask.distributed import progress

result = ddf.groupby("customer_id")["amount"].sum()
future = client.compute(result)
progress(future)
df = future.result()

Variantes

Dask con S3

ddf = dd.read_parquet(
    "s3://my-bucket/data/orders/",
    storage_options={"key": "aws-key", "secret": "aws-secret"},
)

ddf.to_parquet(
    "s3://my-bucket/data/output/",
    storage_options={"key": "aws-key", "secret": "aws-secret"},
)

Dask Bag para Data No Estructurada

import dask.bag as db

bag = db.read_text("data/events_*.jsonl").map(json.loads)

result = (
    bag
    .filter(lambda x: x["event_type"] == "purchase")
    .map(lambda x: {"user": x["user_id"], "amount": x["amount"]})
    .to_dataframe()
    .compute()
)

Dask Array para Cargas de NumPy

import dask.array as da

x = da.random.random((10000, 10000), chunks=(1000, 1000))
mean = x.mean(axis=0)
result = mean.compute()

Mejores Prácticas

Para trabajo local, yo empiezo con npartitions alrededor de dos o cuatro veces el número de cores. Eso da suficiente paralelismo sin generar demasiado overhead de scheduling. Apuntá a particiones de 50 MB a 200 MB. Las chicas agregan overhead; las grandes dejan cores ociosos.

Llamá a .compute() solo al final de una cadena para que Dask optimice el task graph completo. Usá .persist() cuando vayas a tocar un DataFrame más de una vez; así se mantiene en memoria de los workers.

Prefiero Parquet sobre CSV porque conserva tipos, soporta column pruning y se lee más rápido. Para operaciones que la API de Dask no expone directamente, bajá a map_partitions y escribí código normal de pandas en cada chunk. Evitá .set_index() en DataFrames grandes porque dispara un shuffle completo. Para desarrollo local, usá el Dask Distributed scheduler; agrega un dashboard y mejores diagnósticos que el scheduler síncrono por defecto.

Errores Comunes

Llamar a .compute() muy temprano materializa resultados intermedios y rompe la optimización del graph. Encadená operaciones y llamá .compute() una sola vez al final. Demasiadas particiones, como 1.000 particiones chicas de 1 MB, generan un overhead de scheduling enorme. Llamá a .repartition() para caer en el rango de 50–200 MB.

No usar .persist() para datos reusados significa que Dask recomputa el task graph cada vez. Dejá los objetos que vas a reusar en memoria con .persist(). CSV es más lento que Parquet porque debe parsear tipos en cada lectura y no soporta column pruning. Además, Dask lee una muestra para inferir el dtype del CSV, lo cual puede ser incorrecto. Declará los dtypes para evitar errores de tipo.

Resumen

Dask escala workflows de pandas partiendo DataFrames en particiones y construyendo un task graph lazy que se ejecuta en paralelo. Usalo cuando los datasets sean demasiado grandes para la memoria pero no justifiquen un cluster de Spark. Usá Parquet para I/O, llamá .compute() una sola vez al final de una cadena, y .persist() cuando vayas a reusar un DataFrame. Empezá con dos a cuatro veces el número de cores para npartitions, apuntá a 50–200 MB por partición, y bajá a map_partitions o delayed cuando la API high-level no exponga lo que necesitás. El scheduler distribuido agrega un dashboard y mejores diagnósticos incluso en una sola máquina, así que usalo también para trabajo local.

See Also

La documentación de Dask DataFrame cubre la API completa y buenas prácticas en profundidad. Para el scheduler distribuido, la documentación de Dask distributed explica deployment, workers y diagnósticos. La especificación del formato Apache Parquet explica por qué Parquet supera a CSV para workloads analíticos. Si estás validando datos antes de cargarlos en Dask, la receta de validación con Pandera complementa bien este pipeline. Para orquestar jobs de Dask en un schedule, la receta de scheduling con Airflow DAG muestra cómo conectar Dask con un workflow manager.