Impulsa tu flujo de trabajo en ciencia de datos con Dask: guía completa

Cuando Wes McKinney empezó a crear pandas, estableció una regla práctica: para que pandas funcione de forma óptima, la RAM de la máquina debe ser entre 5 y 10 veces mayor que el tamaño del dataset en cuestión. Esta regla era fácil de seguir en 2010, pero ahora estamos en 2023.

Para 2020, los datasets del mundo real ya habían crecido hasta tamaños que podían bloquear sin esfuerzo portátiles y equipos corrientes. Anticipando este problema con tiempo, en 2015 se lanzó una solución.

Dask es una biblioteca de código abierto desarrollada por los creadores de Anaconda para abordar los retos de la computación escalable y eficiente con grandes volúmenes de datos que exceden la memoria de una sola máquina.

Este tutorial ofrece una introducción completa a Dask y a sus funciones clave, incluidas sus interfaces para DataFrames, Arrays y Bags.

Configurar Dask

Como cualquier otra biblioteca, puedes instalar Dask de tres formas: con Conda, con Pip y desde el código fuente.

Como este es un artículo introductorio sobre Dask, no cubriremos el último método de instalación, pensado para mantenedores.

Si usas Anaconda, Dask viene incluido en la instalación por defecto (lo que dice mucho de su popularidad). Si quieres reinstalarlo o actualizarlo, puedes usar el comando install:

conda install dask

La alternativa con PIP es la siguiente:

pip install "dask[complete]"

Añadir la extensión [complete] también instala las dependencias requeridas por Dask, evitando tener que instalar manualmente NumPy, pandas y Tornado.

Puedes comprobar si la instalación fue correcta mirando la versión de la biblioteca:

import dask

dask.__version__

Salida:

'2023.5.0'

La mayor parte de tu trabajo con Dask se centrará en tres interfaces: Dask DataFrames, Arrays y Bags. Vamos a importarlas junto con numpy y pandas para el resto del artículo:

import dask.array as da
import dask.bag as db
import dask.dataframe as dd
import numpy as np
import pandas as pd

Conceptos básicos de Dask

A alto nivel, puedes pensar en Dask como un envoltorio que amplía las capacidades de herramientas tradicionales como pandas, NumPy y Spark para manejar conjuntos de datos más grandes que la memoria.

Cuando se enfrenta a objetos grandes como arrays (vectores) o matrices (dataframes) que no caben en memoria, Dask los divide en bloques, también llamados particiones.

Por ejemplo, considera el array de 12 números aleatorios en NumPy y en Dask:

narr = np.random.rand(12)

narr

array([0.9261154 , 0.87774082, 0.87078873, 0.22309476, 0.24575174,
      0.04182393, 0.31476305, 0.04599283, 0.62354124, 0.97597454,
      0.23923457, 0.81201211])

darr = da.from_array(narr, chunks=3)
darr

La imagen anterior muestra que el array de Dask contiene cuatro bloques porque hemos establecido chunks en 3. Por debajo, cada bloque es en sí mismo un array de NumPy.

Ahora, veamos un ejemplo mucho mayor. Crearemos dos arrays de 10k por 100k (1.000 millones de elementos) y realizaremos una multiplicación elemento a elemento en ambas bibliotecas mientras medimos el rendimiento:

# Create the NumPy arrays
arr1 = np.random.rand(10_000, 100_000)
arr2 = np.random.rand(10_000, 100_000)

# Create the Dask arrays
darr1 = da.from_array(arr1, chunks=(1_000, 10_000))
darr2 = da.from_array(arr2, chunks=(1_000, 10_000))

%%time

result_np = np.multiply(arr1, arr2)

Wall time: 3.19 s

%%time

result_dask = da.multiply(darr1, darr2)

Wall time: 94.8 ms

El código anterior muestra la multiplicación elemento a elemento de dos arrays grandes tanto con NumPy como con Dask. Como ves en la salida, Dask es aproximadamente 34 veces más rápido que NumPy para este cálculo. Las mejoras de rendimiento son aún mayores a medida que aumentan la complejidad del cálculo y el tamaño del array.

Dask utiliza un enfoque similar de dividir en bloques y distribuirlos entre todos los núcleos disponibles de tu máquina para otros tipos de objetos.

Dask DataFrames

Para apreciar de verdad las ventajas de Dask, necesitamos un dataset grande, preferiblemente de más de 1 GB. Sin embargo, descargar un dataset así para seguir el tutorial quizá no sea lo ideal. En su lugar, puedes usar este script, que genera un conjunto de datos sintético con 10 millones de filas, 10 variables numéricas y 10 categóricas.

Asegúrate de que tu equipo tenga al menos 12 GB de RAM para ejecutar el script.

Una vez tengas el archivo large_dataset.csv en tu espacio de trabajo, puedes cargarlo con la función read_csv de la interfaz de Dask DataFrames (dd):

import dask.dataframe as dd

dask_df = dd.read_csv("data/large_dataset.csv")

dask_df.head()

Aunque el archivo es grande, verás que el resultado se obtiene casi al instante. Para archivos aún mayores, puedes especificar el parámetro `blocksize`, que determina en cuántos bytes se trocea el archivo.

Igual que los Dask Arrays contienen bloques de pequeños arrays de NumPy, Dask está diseñado para gestionar múltiples DataFrames de pandas pequeños dispuestos a lo largo del índice de filas.

Como podrás intuir por la función read_csv, la mayor parte de la sintaxis y funcionalidades más usadas del API de pandas se conservan en Dask. Los siguientes bloques de código te resultarán familiares si ya has trabajado con pandas.

Seleccionar columnas y operaciones elemento a elemento

En este ejemplo, hacemos operaciones bastante directas con columnas en nuestro Dask DataFrame, llamado dask_df. Sumamos los valores de la columna Numeric_0 al resultado de multiplicar los valores de Numeric_9 y Numeric_3. Guardamos el resultado en una variable llamada result.

result = (
   dask_df["Numeric_0"] + dask_df["Numeric_9"] * dask_df["Numeric_3"]
)

result.compute().head()


0    1.301960
1    1.190679
2    1.100955
3    0.758272
4    0.926729
dtype: float64

Como hemos comentado, Dask es un poco diferente de las herramientas tradicionales porque no ejecuta estas operaciones al momento. En su lugar, crea una especie de «plan» llamado grafo de tareas para llevarlas a cabo más adelante. Este enfoque permite a Dask optimizar y paralelizar los cálculos cuando hace falta. La función compute() es la que desencadena finalmente la ejecución, y head() nos muestra las primeras filas del resultado.

Filtrado condicional

Ahora veamos cómo filtrar datos con Dask. Estamos seleccionando filas de nuestro DataFrame donde el valor en la columna "Categorical_5" es "A".

Este filtrado es similar a cómo lo harías en pandas, pero con un matiz: Dask realiza la operación de forma perezosa. Prepara el grafo de tareas, pero espera a ejecutarlo hasta que llamamos a compute(). Al ejecutar head(), vemos las primeras filas del DataFrame filtrado.

dask_df[dask_df["Categorical_5"] == "A"].compute().head()

Estadísticos descriptivos comunes

A continuación, vamos a generar algunos estadísticos descriptivos habituales con la función describe() de Dask.

Nos devuelve varias métricas descriptivas para nuestro DataFrame, como la media, la desviación estándar, el mínimo, el máximo, etc. Como en los ejemplos anteriores, Dask prepara el grafo de tareas cuando llamamos a describe(), pero espera a ejecutarlo hasta que invoquemos compute().

dask_df.describe().compute()

dask_df["Categorical_3"].value_counts().compute().head()

Categorical_3
O    386038
C    385804
A    385493
P    385490
K    385116
Name: count, dtype: int64

También usamos value_counts() para contar cuántas veces aparece cada valor único en la columna "Categorical_3". Disparamos la operación con compute(), y head() nos muestra los valores más frecuentes.

Groupby

Por último, usemos la función groupby() para agrupar los datos según los valores de la columna "Categorical_8". Después, seleccionamos la columna "Numeric_7" y calculamos la media para cada grupo.

Esto es similar a cómo usarías groupby() en pandas, pero, como ya imaginarás, Dask lo hace de forma perezosa. Ejecutamos la operación con compute(), y head() muestra la media de la columna "Numeric_7" para los primeros grupos.

dask_df.groupby("Categorical_8")["Numeric_7"].mean().compute().head()


Categorical_8
A    0.498497
B    0.499767
C    0.500622
D    0.500307
E    0.499530
Name: Numeric_7, dtype: float64

Echa un vistazo a esta sección de la guía de usuario de Dask para ver más similitudes entre pandas y Dask.

Evaluación perezosa

Ahora, vamos a profundizar en el uso de la función compute al final de cada bloque de código.

Dask evalúa los bloques en modo perezoso, a diferencia del modo ansioso de pandas, que devuelve resultados de inmediato.

Si lo llevamos a la cocina, la evaluación perezosa es como preparar ingredientes y cortar verduras por adelantado, pero combinarlo todo para cocinar solo cuando hace falta. Esa es la función de compute.

En cambio, la evaluación ansiosa sería ir echando los ingredientes al fuego en cuanto están listos, asegurando que todo termine a la vez para servir.

La evaluación perezosa es clave para el excelente rendimiento de Dask porque aporta:

1. Menos cómputo. Las expresiones se evalúan solo cuando se necesitan (cuando se llama a compute), evitando resultados intermedios innecesarios que quizá no se usen en el resultado final.
2. Asignación óptima de recursos. Evita dedicar memoria o CPU a resultados intermedios que puede que no hagan falta.
3. Soporte para grandes volúmenes de datos. Este método procesa los datos sobre la marcha o en bloques más pequeños, utilizando la memoria de forma eficiente.

Cuando se devuelven los resultados de compute, se entregan como Series/DataFrames de pandas o arrays de NumPy, en lugar de como objetos nativos de Dask.

>>> type(dask_df)
dask.dataframe.core.DataFrame


>>> type(
   dask_df[["Numeric_5", "Numeric_6", "Numeric_7"]].mean().compute()
)

pandas.core.series.Series

La razón es que la mayoría de operaciones de manipulación de datos devuelven solo un subconjunto del dataframe original, ocupando mucho menos espacio. Así que no suele ser necesario seguir usando el paralelismo de Dask, y puedes continuar el resto del flujo de trabajo en pandas o NumPy.

Dask Bags y Dask Delayed para datos no estructurados

Dask Bags y Dask Delayed son dos componentes de la biblioteca Dask que proporcionan potentes herramientas para trabajar con datos no estructurados o semiestructurados y habilitar la evaluación perezosa.

Mientras que antes los datos tabulares eran los más comunes, hoy en día los datasets suelen incluir archivos no estructurados como imágenes, textos, vídeos y audio. Dask Bags ofrece la funcionalidad y el API para manejar este tipo de archivos de forma paralela y escalable.

Por ejemplo, considera esta sencilla ilustración:

import dask.bag as db

# Create a Dask Bag from a list of strings
b = db.from_sequence(["apple", "banana", "orange", "grape", "kiwi"])

# Filter the strings that start with the letter 'a'
filtered_strings = b.filter(lambda x: x.startswith("a"))

# Map a function to convert each string to uppercase
uppercase_strings = filtered_strings.map(lambda x: x.upper())

# Compute the result as a list
result = uppercase_strings.compute()

print(result)
['APPLE']

En este ejemplo, creamos un Dask Bag b a partir de una lista de strings. Después aplicamos operaciones sobre el Bag para filtrar las cadenas que empiezan por la letra "a" y convertirlas a mayúsculas con las funciones filter() y map(), respectivamente. Por último, calculamos el resultado como una lista con compute() e imprimimos la salida.

Ahora imagina que puedes realizar operaciones mucho más complejas sobre miles de millones de cadenas similares almacenadas en un archivo de texto. Sin la evaluación perezosa y el paralelismo que ofrece Dask Bags, te enfrentarías a retos considerables. (Lee más sobre Bags en la documentación de Dask).

En cuanto a Dask Delayed, aporta aún más flexibilidad e introduce evaluación perezosa y paralelismo en muchos otros escenarios. Con Dask Delayed, puedes convertir cualquier función nativa de Python en un objeto perezoso usando el decorador @dask.delayed.

Aquí tienes un ejemplo sencillo:

%%time

import time
import dask


@dask.delayed
def process_data(x):
   # Simulate some computation
   time.sleep(1)
   return x**2


# Generate a list of inputs
inputs = range(1000)

# Apply the delayed function to each input
results = [process_data(x) for x in inputs]

# Compute the results in parallel
computed_results = dask.compute(*results)



Wall time: 42.1 s

En este ejemplo, definimos una función process_data decorada con @dask.delayed. La función simula trabajo computacional durmiendo 1 segundo y después devuelve el cuadrado del valor de entrada.

Sin paralelismo, realizar este cómputo para 1000 entradas habría tardado más de 1000 segundos. Sin embargo, con Dask Delayed y la ejecución en paralelo, el cálculo solo tardó unos 42,1 segundos.

Este ejemplo demuestra la potencia del paralelismo para reducir el tiempo de cómputo distribuyendo la carga de trabajo de forma eficiente entre múltiples núcleos o workers.

De eso va el paralelismo.

Conclusión y recursos adicionales

Dask es una de las bibliotecas clave del ecosistema de datos. Amplía la funcionalidad de gran parte de las bibliotecas más usadas, como NumPy, pandas y Spark, permitiendo trabajar sin problemas con datasets que superan la memoria.

Con Dask Bags y Dask Delayed, lleva el paralelismo y la evaluación perezosa a escenarios menos tradicionales, como trabajar con datos no estructurados o con objetos nativos de Python.

Para los detalles más finos, no dejes de leer a fondo la documentación de Dask.

Si buscas un recurso completo para dominar Dask, échale un vistazo a nuestro curso, Parallel Programming With Dask in Python.