¿Quién soy yo?¶
- Ingeniero Aeron谩utico y pythonista autodidacta
- Mission Planning & Execution Engineer en Satellogic
- Presidente de la Asociaci贸n Python Espa帽a 馃悕 馃嚜馃嚫
- Colaborador en proyectos de Python Cient铆fico: NumPy, SciPy, conda, astropy, memory-profiler...
- Profesor asociado en el Instituto Empresa de Python para Big Data
- Amante de la pizza y el hard rock 馃
Resumen¶
- Estado de la cuesti贸n
- Dask
- Introducci贸n
- Evaluaci贸n perezosa
- Grafos de operaciones
- Demo
- Limitaciones
- Proyectos relacionados
- Conclusiones y futuro
¿El principio del fin de PySpark?¶
2. Dask¶
dask es una biblioteca de computaci贸n paralela orientada a la anal铆tica. Est谩 formada por dos componentes:
- Dynamic task scheduling optimizada para la computaci贸n.
- Colecciones "Big Data" como arrays, DataFrames y listas paralelas, que mimetizan la forma de trabajar con NumPy, pandas o iteradores de Python para objetos m谩s grandes que la memoria disponible o en entornos distribuidos. Estas colecciones funcionan sobre los schedulers.
Es un proyecto joven pero tiene determinadas propiedades que lo hacen muy interesante, entre ellas:
- Familiar: Dask replica la forma de trabajar con arrays de Numpy y DataFrames de pandas, as铆 que la transici贸n es mucho m谩s sencilla que con otros sistemas.
- Flexible: Se integra bien con otros proyectos y provee herramientas para paralelizar nuestras propias funciones.
- Nativo: Es Python puro, no hay antipatrones ni comunicaci贸n con otros lenguajes.
- Escalable: Dask funciona tanto en clusters de 1000 nodos como en port谩tiles normales, optimizando el uso de memoria.
- Amistoso: Proporciona feedback inmediato y abundantes herramientas de diagn贸stico.
Instalación¶
La versi贸n m谩s reciente es la 2.9.0 (2019-12-06, 隆hace unos d铆as!) y se puede instalar con pip:
$ pip install dask[complete]o con conda:
$ conda install daskEvaluación perezosa¶
Vamos a hacer un ejemplo trivial con dask.array para comprobar c贸mo funciona la computaci贸n en dask.
import numpy as np
import dask.array as da
x = np.arange(1000)
y = da.from_array(x, chunks=100)
y
Si intentamos efectuar cualquier operaci贸n sobre estos arrays, no se ejecuta inmediatamente:
op = y.mean()
op
Dask en su lugar construye un grafo con todas las operaciones necesarias y sus dependencias para que podamos visualizarlo y razonar sobre 茅l. Este grafo est谩 almacenado en estructuras de datos corrientes de Python como diccionarios, listas y tuplas:
y.dask.dicts
Y podemos visualizarlo si tenemos instalada la biblioteca graphviz:
op.visualize()
Si queremos efectuar la operaci贸n, tendremos que llamar al m茅todo .compute().
op.compute()
Si queremos convertir nuestro array original a array de NumPy, tambi茅n se hace llamando a compute():
y.compute()
Demo: Taxis de NYC¶
Otra de las estructuras de datos que provee pandas son los DataFrames, que se comportan de la misma manera que los DataFrames de pandas.
Para estudiar c贸mo funciona, vamos a descargar datos de trayectos de los taxis de New York:
!du data/yellow*.csv -h -s
!du data/ -h -s
Tanto dask.dataframe como dask.array usan un scheduler por defecto basado en hilos. En su lugar, vamos a utilizar una clase Client, la que emplear铆amos si estuvi茅ramos en un cluster.
import dask.dataframe as dd
Esta clase Client, cuando se utiliza en local, lanza un scheduler que minimiza el uso de memoria y aprovecha todos los n煤cleos de la CPU.
"The dask single-machine schedulers have logic to execute the graph in a way that minimizes memory footprint." http://dask.pydata.org/en/latest/custom-graphs.html?highlight=minimizes%20memory#related-projects
El servidor de diagn贸stico est谩 disponible en http://127.0.0.1:8787/.
from dask.distributed import Client
client = Client()
client
Y ahora leemos los .csv con un filtro todos a la vez en el mismo DataFrame de Dask:
df = dd.read_csv("data/yellow*.csv",
parse_dates=['tpep_pickup_datetime', 'tpep_dropoff_datetime'])
Que mimetiza la API de pandas:
df.head()
df.dtypes
Vamos a calcular la longitud del DataFrame:
# Esta operaci贸n bloquea el int茅rprete durante unos minutos
len(df)
Como se puede observar, el uso de memoria est谩 contenido y todas las CPUs est谩n trabajando.
Tambi茅n lo podemos hacer de manera as铆ncrona:
futures = client.submit(len, df)
futures
from distributed import progress
Vamos ahora a calcular la distancia media recorrida en funci贸n del n煤mero de ocupantes. Igual que cuando us谩bamos dask.array, la operaci贸n no se efect煤a autom谩ticamente.
op = df.groupby(df.passenger_count).trip_distance.mean()
op
f2 = client.compute(op)
f2
client.compute almacena el resultado en un solo nodo, y por tanto debe usarse con cuidado. Para objetos grandes, es mejor usar client.persist.progress(f2)
f2.result()
En este caso la visualizaci贸n de la operaci贸n ya tiene una magnitud considerable:
op.visualize()