dask

¿Qué hace dask?

Resumen

Dask es una biblioteca de Python para cómputo paralelo y distribuido que ofrece tres capacidades críticas:

• Ejecución más grande que la memoria en una sola máquina para datos que exceden la RAM disponible
• Procesamiento paralelo para mejorar la velocidad de cómputo en múltiples núcleos
• Cómputo distribuido que soporta conjuntos de datos a escala de terabytes en múltiples máquinas

Dask escala desde laptops (procesando ~100 GiB) hasta clústeres (procesando ~100 TiB) manteniendo APIs familiares de Python.

Versión actual ascendente: dask 2026.3.0 (PyPI, marzo 2026). Documentación: docs.dask.org. Desde 2025.1.0, la API de DataFrame basada en expresiones con planificación de consultas es la única implementación — no instales dask-expr por separado ni establezcas dataframe.query-planning: False.

Inicio Rápido

Instalación

uv pip install "dask>=2025.1"

Para un flujo de trabajo típico con pandas/NumPy que incluya el planificador distribuido y el panel de control:

uv pip install "dask[complete]"

Almacenamiento remoto de objetos (S3, GCS, Azure):

uv pip install s3fs    # rutas s3://
uv pip install gcsfs   # rutas gs://

Requiere Python 3.10+ (el soporte para 3.9 se eliminó en 2024.12). La E/S de DataFrame requiere PyArrow 16+ (a partir de dask 2026.1.2).

Cuándo Usar Esta Habilidad

Esta habilidad debe usarse cuando:

• Procesas conjuntos de datos que exceden la RAM disponible
• Escalas operaciones de pandas o NumPy a conjuntos de datos más grandes
• Paralelizas cómputos para mejorar el rendimiento
• Procesas múltiples archivos de manera eficiente (CSV, Parquet, JSON, registros de texto)
• Construyes flujos de trabajo paralelos personalizados con dependencias de tareas
• Distribuyes cargas de trabajo en múltiples núcleos o máquinas

Capacidades Principales

Dask proporciona cinco componentes principales, cada uno adecuado para diferentes casos de uso:

1. DataFrames - Operaciones Paralelas de Pandas

Propósito: Escalar operaciones de pandas a conjuntos de datos más grandes mediante procesamiento paralelo.

Cuándo Usar:

• Los datos tabulares exceden la RAM disponible
• Necesitas procesar múltiples archivos CSV/Parquet juntos
• Las operaciones de pandas son lentas y necesitan paralelización
• Escalar un prototipo de pandas a producción

Documentación de Referencia: Para una guía completa sobre Dask DataFrames, consulta references/dataframes.md, que incluye:

• Lectura de datos (archivo único, múltiples archivos, patrones glob)
• Operaciones comunes (filtrado, groupby, joins, agregaciones)
• Operaciones personalizadas con map_partitions
• Consejos de optimización de rendimiento
• Patrones comunes (ETL, series temporales, procesamiento de múltiples archivos)

Ejemplo Rápido:

import dask.dataframe as dd

# Leer múltiples archivos como un solo DataFrame
ddf = dd.read_csv('data/2024-*.csv')

# Las operaciones son perezosas hasta compute()
filtered = ddf[ddf['value'] > 100]
result = filtered.groupby('category').mean().compute()

Puntos Clave:

• Las operaciones son perezosas (construyen un grafo de tareas) hasta que se llama a .compute()
• Usa map_partitions para operaciones personalizadas eficientes
• Convierte a DataFrame temprano cuando trabajes con datos estructurados de otras fuentes

2. Arrays - Operaciones Paralelas de NumPy

Propósito: Extender las capacidades de NumPy a conjuntos de datos más grandes que la memoria usando algoritmos por bloques.

Cuándo Usar:

• Los arrays exceden la RAM disponible
• Las operaciones de NumPy necesitan paralelización
• Trabajas con conjuntos de datos científicos (HDF5, Zarr, NetCDF)
• Necesitas álgebra lineal paralela u operaciones con arrays

Documentación de Referencia: Para una guía completa sobre Dask Arrays, consulta references/arrays.md, que incluye:

• Creación de arrays (desde NumPy, aleatorios, desde disco)
• Estrategias de fragmentación y optimización
• Operaciones comunes (aritméticas, reducciones, álgebra lineal)
• Operaciones personalizadas con map_blocks
• Integración con HDF5, Zarr y XArray

Ejemplo Rápido:

import dask.array as da

# Crear un array grande con fragmentos
x = da.random.random((100000, 100000), chunks=(10000, 10000))

# Las operaciones son perezosas
y = x + 100
z = y.mean(axis=0)

# Calcular resultado
result = z.compute()

Puntos Clave:

• El tamaño del fragmento es crítico (apunta a ~100 MB por fragmento)
• Las operaciones trabajan en fragmentos en paralelo
• Re-fragmenta los datos cuando sea necesario para operaciones eficientes
• Usa map_blocks para operaciones no disponibles en Dask

3. Bags - Procesamiento Paralelo de Datos No Estructurados

Propósito: Procesar datos no estructurados o semiestructurados (texto, JSON, registros) con operaciones funcionales.

Cuándo Usar:

• Procesar archivos de texto, registros o registros JSON
• Limpieza de datos y ETL antes del análisis estructurado
• Trabajar con objetos de Python que no se ajustan a formatos de array/DataFrame
• Necesitas procesamiento en flujo eficiente en memoria

Documentación de Referencia: Para una guía completa sobre Dask Bags, consulta references/bags.md, que incluye:

• Lectura de archivos de texto y JSON
• Operaciones funcionales (map, filter, fold, groupby)
• Conversión a DataFrames
• Patrones comunes (análisis de registros, procesamiento JSON, procesamiento de texto)
• Consideraciones de rendimiento

Ejemplo Rápido:

import dask.bag as db
import json

# Leer y analizar archivos JSON
bag = db.read_text('logs/*.json').map(json.loads)

# Filtrar y transformar
valid = bag.filter(lambda x: x['status'] == 'valid')
processed = valid.map(lambda x: {'id': x['id'], 'value': x['value']})

# Convertir a DataFrame para análisis
ddf = processed.to_dataframe()

Puntos Clave:

• Úsalo para limpieza inicial de datos, luego convierte a DataFrame/Array
• Usa foldby en lugar de groupby para mejor rendimiento
• Las operaciones son en flujo y eficientes en memoria
• Convierte a formatos estructurados (DataFrame) para operaciones complejas

4. Futures - Paralelización Basada en Tareas

Propósito: Construir flujos de trabajo paralelos personalizados con control detallado sobre la ejecución de tareas y dependencias.

Cuándo Usar:

• Construir flujos de trabajo dinámicos y en evolución
• Necesitas ejecución inmediata de tareas (no perezosa)
• Los cómputos dependen de condiciones de tiempo de ejecución
• Implementar algoritmos paralelos personalizados
• Necesitas cómputos con estado

Documentación de Referencia: Para una guía completa sobre Dask Futures, consulta references/futures.md, que incluye:

• Configuración del cliente distribuido
• Envío de tareas y trabajo con futures
• Dependencias de tareas y movimiento de datos
• Coordinación avanzada (colas, bloqueos, eventos, actores)
• Patrones comunes (barridos de parámetros, tareas dinámicas, algoritmos iterativos)

Ejemplo Rápido:

from dask.distributed import Client

client = Client()  # Crear clúster local

# Enviar tareas (se ejecutan inmediatamente)
def process(x):
    return x ** 2

futures = client.map(process, range(100))

# Recopilar resultados
results = client.gather(futures)

client.close()

Puntos Clave:

• Requiere un cliente distribuido (incluso para una sola máquina)
• Las tareas se ejecutan inmediatamente cuando se envían
• Pre-dispersa datos grandes para evitar transferencias repetidas
• ~1ms de sobrecarga por tarea (no adecuado para millones de tareas pequeñas)
• Usa actores para flujos de trabajo con estado

5. Planificadores - Motores de Ejecución

Propósito: Controlar cómo y dónde se ejecutan las tareas de Dask (hilos, procesos, distribuido).

Cuándo Elegir un Planificador:

• Hilos (predeterminado): Operaciones NumPy/Pandas, bibliotecas que liberan el GIL, beneficio de memoria compartida
• Procesos: Código Python puro, procesamiento de texto, operaciones limitadas por el GIL
• Síncrono: Depuración con pdb, perfilado, comprensión de errores
• Distribuido: Necesitas panel de control, clústeres multimáquina, funciones avanzadas

Documentación de Referencia: Para una guía completa sobre los Planificadores de Dask, consulta references/schedulers.md, que incluye:

• Descripciones detalladas de los planificadores y sus características
• Métodos de configuración (global, administrador de contexto, por cómputo)
• Consideraciones de rendimiento y sobrecarga
• Patrones comunes y solución de problemas
• Configuración de hilos para rendimiento óptimo

Ejemplo Rápido:

import dask
import dask.dataframe as dd

# Usar hilos para DataFrame (predeterminado, bueno para numérico)
ddf = dd.read_csv('data.csv')
result1 = ddf.mean().compute()  # Usa hilos

# Usar procesos para trabajo pesado en Python
import dask.bag as db
bag = db.read_text('logs/*.txt')
result2 = bag.map(python_function).compute(scheduler='processes')

# Usar síncrono para depuración
dask.config.set(scheduler='synchronous')
result3 = problematic_computation.compute()  # Puedes usar pdb

# Usar distribuido para monitoreo y escalado
from dask.distributed import Client
client = Client()
result4 = computation.compute()  # Usa distribuido con panel de control

Puntos Clave:

• Hilos: Menor sobrecarga (~10 µs/tarea), mejor para trabajo numérico
• Procesos: Evita el GIL (~10 ms/tarea), mejor para trabajo en Python
• Distribuido: Panel de control de monitoreo (~1 ms/tarea), escala a clústeres
• Puedes cambiar de planificador por cómputo o globalmente

Mejores Prácticas

Para una guía completa sobre optimización de rendimiento, estrategias de gestión de memoria y errores comunes a evitar, consulta references/best-practices.md. Los principios clave incluyen:

Comienza con Soluciones Más Simples

Antes de usar Dask, explora:

• Mejores algoritmos
• Formatos de archivo eficientes (Parquet en lugar de CSV)
• Código compilado (Numba, Cython)
• Muestreo de datos

Reglas Críticas de Rendimiento

1. No Cargues Datos Localmente y Luego Pásalos a Dask

# Incorrecto: Carga todos los datos en memoria primero
import pandas as pd
df = pd.read_csv('large.csv')
ddf = dd.from_pandas(df, npartitions=10)

# Correcto: Deja que Dask maneje la carga
import dask.dataframe as dd
ddf = dd.read_csv('large.csv')

2. Evita Llamadas Repetidas a compute()

# Incorrecto: Cada compute es separado
for item in items:
    result = dask_computation(item).compute()

# Correcto: Un solo compute para todos
computations = [dask_computation(item) for item in items]
results = dask.compute(*computations)

3. No Construyas Grafos de Tareas Excesivamente Grandes

• Aumenta el tamaño de los fragmentos si hay millones de tareas
• Usa map_partitions/map_blocks para fusionar operaciones
• Verifica el tamaño del grafo de tareas: len(ddf.__dask_graph__())

4. Elige Tamaños de Fragmento Apropiados

• Objetivo: ~100 MB por fragmento (o 10 fragmentos por núcleo en la memoria del trabajador)
• Demasiado grande: Desbordamiento de memoria
• Demasiado pequeño: Sobrecarga de planificación

5. Usa el Panel de Control

from dask.distributed import Client
client = Client()
print(client.dashboard_link)  # Monitorear rendimiento, identificar cuellos de botella

Patrones de Flujo de Trabajo Comunes

Tubería ETL

import dask.dataframe as dd

# Extraer: Leer datos
ddf = dd.read_csv('raw_data/*.csv')

# Transformar: Limpiar y procesar
ddf = ddf[ddf['status'] == 'valid']
ddf['amount'] = ddf['amount'].astype('float64')
ddf = ddf.dropna(subset=['important_col'])

# Cargar: Agregar y guardar
summary = ddf.groupby('category').agg({'amount': ['sum', 'mean']})
summary.to_parquet('output/summary.parquet')

Tubería de No Estructurado a Estructurado

import dask.bag as db
import json

# Comenzar con Bag para datos no estructurados
bag = db.read_text('logs/*.json').map(json.loads)
bag = bag.filter(lambda x: x['status'] == 'valid')

# Convertir a DataFrame para análisis estructurado
ddf = bag.to_dataframe()
result = ddf.groupby('category').mean().compute()

Cómputo de Arrays a Gran Escala

import dask.array as da

# Cargar o crear array grande
x = da.from_zarr('large_dataset.zarr')

# Procesar en fragmentos
normalized = (x - x.mean()) / x.std()

# Guardar resultado (usa mode= para sobrescribir; zarr_array_kwargs para compresión)
da.to_zarr(normalized, 'normalized.zarr', mode='w')

Flujo de Trabajo Paralelo Personalizado

from dask.distributed import Client

client = Client()

# Dispersar un conjunto de datos grande una vez
data = client.scatter(large_dataset)

# Procesar en paralelo con dependencias
futures = []
for param in parameters:
    future = client.submit(process, data, param)
    futures.append(future)

# Recopilar resultados
results = client.gather(futures)

Seleccionar el Componente Correcto

Usa esta guía de decisión para elegir el componente adecuado de Dask:

Tipo de Datos:

• Datos tabulares → DataFrames
• Arrays numéricos → Arrays
• Texto/JSON/registros → Bags (luego convertir a DataFrame)
• Objetos Python personalizados → Bags o Futures

Tipo de Operación:

• Operaciones estándar de pandas → DataFrames
• Operaciones estándar de NumPy → Arrays
• Tareas paralelas personalizadas → Futures
• Procesamiento de texto/ETL → Bags

Nivel de Control:

• Alto nivel, automático → DataFrames/Arrays
• Bajo nivel, manual → Futures

Tipo de Flujo de Trabajo:

• Grafo de cómputo estático → DataFrames/Arrays/Bags
• Dinámico, en evolución → Futures

Consideraciones de Integración

Formatos de Archivo

• Eficientes: Parquet, HDF5, Zarr (columnar, comprimidos, amigables con paralelismo)
• Compatibles pero más lentos: CSV (úselo solo para ingesta inicial)
• Para Arrays: HDF5, Zarr, NetCDF

Conversión Entre Colecciones

# Bag → DataFrame
ddf = bag.to_dataframe()

# DataFrame → Array (para datos numéricos)
arr = ddf.to_dask_array(lengths=True)

# Array → DataFrame
ddf = dd.from_dask_array(arr, columns=['col1', 'col2'])

Con Otras Bibliotecas

• XArray: Envuelve arrays de Dask con dimensiones etiquetadas (geoespacial, imágenes)
• Dask-ML: Aprendizaje automático con APIs compatibles con scikit-learn
• Distributed: Gestión avanzada de clústeres y monitoreo

Depuración y Desarrollo

Flujo de Trabajo de Desarrollo Iterativo

1. Prueba con datos pequeños y planificador síncrono:

dask.config.set(scheduler='synchronous')
result = computation.compute()  # Puede usar pdb, depuración fácil

2. Valida con hilos en una muestra:

sample = ddf.head(1000)  # Muestra pequeña
# Prueba la lógica, luego escala al conjunto de datos completo

3. Escala con distribuido para monitoreo:

from dask.distributed import Client
client = Client()
print(client.dashboard_link)  # Monitorear rendimiento
result = computation.compute()

Problemas Comunes

Errores de Memoria:

• Disminuye el tamaño de los fragmentos
• Usa persist() estratégicamente y elimínalo cuando termines
• Revisa fugas de memoria en funciones personalizadas

Inicio Lento:

• Grafo de tareas demasiado grande (aumenta el tamaño de los fragmentos)
• Usa map_partitions o map_blocks para reducir tareas

Mala Paralelización:

• Fragmentos demasiado grandes (aumenta el número de particiones)
• Usar hilos con código Python (cambia a procesos)
• Dependencias de datos que impiden el paralelismo

Sobrecarga de Planificación:

• Demasiadas tareas pequeñas (aumenta el tamaño de los fragmentos)
• Usa dask.compute(*tasks) en lugar de bucles

Consejos Avanzados

Trabajar con Múltiples Fuentes de Datos

# Combinar archivos desde diferentes ubicaciones
import dask.dataframe as dd

ddf1 = dd.read_parquet('s3://bucket/data/*.parquet', storage_options={'anon': True})
ddf2 = dd.read_csv('local/data/*.csv')

# Concatenar
merged = dd.concat([ddf1, ddf2])

Uso Eficiente de Memoria

# Cargar solo columnas necesarias
ddf = dd.read_parquet('data.parquet', columns=['col1', 'col2'])

# Usar tipos de datos más pequeños si es posible
ddf = ddf.astype({'large_col': 'float32'})

Monitoreo de Progreso

from dask.distributed import progress

# En un cuaderno de Jupyter o script
future = client.submit(long_computation, data)
progress(future)  # Mostrar barra de progreso

# O para colecciones perezosas
import dask
futures = dask.persist(*lazy_collections)
progress(futures)

Estrategias de Persistencia

# Mantener datos en memoria del trabajador para acceso repetido
# Útil para datos de referencia o cuando el costo de recarga es alto

ddf_persisted = ddf.persist()
# Ahora las operaciones posteriores son más rápidas

# Liberar memoria cuando ya no se necesita
del ddf_persisted

Referencias

• Documentación de Dask
• Panel de control de Dask
• Repositorio de GitHub de Dask
• Dask-ML
• XArray con Dask