Potenzia il tuo workflow di data science con Dask: guida completa

Quando Wes McKinney iniziò a scrivere pandas, aveva una regola pratica: perché pandas funzioni in modo ottimale, la RAM della macchina deve essere da 5 a 10 volte più grande del dataset in questione. Questa regola era facile da seguire intorno al 2010, ma ora siamo nel 2023.

Già nel 2020, i dataset reali erano cresciuti a dimensioni in grado di mandare in crash con facilità i laptop e le macchine di tutti i giorni. Prevedendo con largo anticipo questo problema, nel 2015 è stata rilasciata una soluzione.

Dask è una libreria open-source sviluppata dai creatori di Anaconda per affrontare le sfide del computing scalabile ed efficiente su grandi dataset che superano la capacità di memoria di una singola macchina.

Questo tutorial offre un’introduzione completa a Dask e alle sue funzionalità fondamentali, incluse le interfacce per DataFrame, Array e Bag.

Configurare Dask

Come qualsiasi altra libreria, puoi installare Dask in tre modi: Conda, Pip e da sorgente.

Dato che questo è un articolo introduttivo su Dask, non tratteremo l’ultimo metodo di installazione, che è pensato per i maintainer.

Se usi Anaconda, Dask è incluso nell’installazione predefinita (un segno di quanto sia popolare la libreria). Se desideri reinstallarlo o aggiornarlo, puoi usare il comando install:

conda install dask

L’alternativa con PIP è la seguente:

pip install "dask[complete]"

Aggiungendo l’estensione [complete] si installano anche le dipendenze richieste da Dask, eliminando la necessità di installare manualmente NumPy, Pandas e Tornado.

Puoi verificare se l’installazione è andata a buon fine controllando la versione della libreria:

import dask

dask.__version__

Output:

'2023.5.0'

La maggior parte del lavoro con Dask sarà incentrata su tre interfacce: Dask DataFrame, Array e Bag. Importiamole insieme a numpy e pandas da usare per il resto dell’articolo:

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

Concetti di base di Dask

A livello alto, puoi pensare a Dask come a un wrapper che estende le capacità di strumenti tradizionali come pandas, NumPy e Spark per gestire dataset più grandi della memoria.

Quando si affrontano oggetti grandi come array (vettori) più grandi della memoria o matrici (dataframe), Dask li suddivide in blocchi, detti anche partizioni.

Ad esempio, considera l’array di 12 numeri casuali sia in NumPy che in 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

L’immagine sopra mostra che l’array Dask contiene quattro chunk avendo impostato chunks a 3. Sotto il cofano, ogni chunk è a sua volta un array NumPy.

Ora consideriamo un esempio molto più grande. Creeremo due array 10k per 100k (1 miliardo di elementi) ed eseguiremo la moltiplicazione elemento per elemento in entrambe le librerie, misurandone le prestazioni:

# 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

Il codice sopra mostra la moltiplicazione elemento per elemento di due grandi array usando sia NumPy che Dask. Come si vede dall’output, Dask è circa 34 volte più veloce di NumPy per questo calcolo. I guadagni in prestazioni diventano ancora più significativi all’aumentare della complessità del calcolo e della dimensione degli array.

Dask utilizza un approccio simile di suddivisione in chunk e distribuzione di questi chunk su tutti i core disponibili della tua macchina anche per altri oggetti.

Dask DataFrame

Per apprezzare davvero i vantaggi di Dask, ci serve un dataset grande, preferibilmente oltre 1 GB. Tuttavia, scaricare un dataset del genere per seguire il tutorial potrebbe non essere ottimale. In alternativa, puoi usare questo script, che genera un dataset sintetico con 10 milioni di righe, 10 feature numeriche e 10 feature categoriche.

Assicurati che la tua macchina abbia almeno 12 GB di RAM per eseguire lo script.

Una volta che hai il file large_dataset.csv nel tuo workspace, puoi caricarlo usando la funzione read_csv dell’interfaccia Dask DataFrame (dd):

import dask.dataframe as dd

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

dask_df.head()

Anche se il file è grande, noterai che il risultato viene ottenuto quasi istantaneamente. Per file ancora più grandi, puoi specificare il parametro `blocksize`, che determina in quanti byte suddividere il file.

Analogamente a come i Dask Array contengono chunk di piccoli array NumPy, Dask è progettato per gestire più piccoli DataFrame Pandas disposti lungo l’indice delle righe.

Come puoi intuire dalla funzione read_csv, gran parte della sintassi e delle funzionalità più comuni dell’API di pandas è mantenuta in Dask. I seguenti blocchi di codice dovrebbero esserti familiari se hai già lavorato con pandas.

Selezione di colonne e operazioni elemento per elemento

In questo esempio eseguiamo operazioni piuttosto semplici sulle colonne del nostro Dask DataFrame, chiamato dask_df. Sommiamo i valori della colonna Numeric_0 al risultato della moltiplicazione tra i valori di Numeric_9 e Numeric_3. Salviamo l’esito in una variabile chiamata 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

Come abbiamo accennato, Dask è un po’ diverso dagli strumenti di calcolo tradizionali perché non esegue immediatamente queste operazioni. Invece, crea una sorta di “piano” chiamato task graph per eseguirle in seguito. Questo approccio consente a Dask di ottimizzare e parallelizzare i calcoli quando necessario. La funzione compute() ordina a Dask di eseguire finalmente queste computazioni, e head() ci mostra semplicemente le prime righe del risultato.

Filtraggio condizionale

Ora vediamo come Dask può filtrare i dati. Selezioniamo le righe del nostro DataFrame in cui il valore nella colonna "Categorical_5" è "A".

Questo processo di filtraggio è simile a come faresti in pandas, ma con una differenza: Dask esegue questa operazione in modo pigro. Prepara il task graph per l’operazione ma aspetta a eseguirla finché non chiamiamo compute(). Quando eseguiamo head(), vediamo le prime righe del DataFrame filtrato.

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

Statistiche descrittive comuni

Successivamente, generiamo alcune statistiche descrittive comuni usando la funzione describe() di Dask.

Restituisce un insieme di statistiche descrittive per il nostro DataFrame, incluse media, deviazione standard, minimo, massimo e così via. Come nei nostri esempi precedenti, Dask prepara il task graph per questa operazione quando chiamiamo describe(), ma aspetta a eseguirla finché non chiamiamo 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

Usiamo anche value_counts() per contare il numero di occorrenze di ciascun valore unico nella colonna "Categorical_3". Attiviamo l’operazione con compute(), e head() ci mostra i valori più comuni.

Groupby

Infine, usiamo la funzione groupby() per raggruppare i dati in base ai valori nella colonna "Categorical_8". Poi selezioniamo la colonna "Numeric_7" e calcoliamo la media per ciascun gruppo.

Questo è simile a come potresti usare groupby() in pandas, ma, come avrai intuito, Dask lo esegue in modo pigro. Attiviamo l’operazione con compute(), e head() mostra la media della colonna "Numeric_7" per i primi gruppi.

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

Dai un’occhiata a questa sezione della guida utente di Dask per il resto delle somiglianze tra pandas e Dask.

Valutazione pigra

Ora esploriamo l’uso della funzione compute alla fine di ciascun blocco di codice.

Dask valuta i blocchi di codice in modalità pigra rispetto alla modalità eager di Pandas, che restituisce i risultati immediatamente.

Per fare un parallelo con la cucina, la valutazione pigra è come preparare in anticipo gli ingredienti e tagliare le verdure, ma unirli per cucinare solo quando serve. La funzione compute serve proprio a questo.

Al contrario, la valutazione eager è come buttare gli ingredienti nel fuoco per cucinarli non appena sono pronti. Questo approccio assicura che tutto sia pronto da servire in una volta sola.

La valutazione pigra è la chiave dell’eccellente performance di Dask perché offre:

1. Calcolo ridotto. Le espressioni sono valutate solo quando necessario (quando si chiama compute), evitando risultati intermedi inutili che potrebbero non essere usati nel risultato finale.
2. Allocazione ottimale delle risorse. La valutazione pigra evita di allocare memoria o potenza di calcolo a risultati intermedi che potrebbero non servire.
3. Supporto per grandi dataset. Questo metodo elabora gli elementi dei dati al volo o in chunk più piccoli, consentendo un utilizzo efficiente della memoria.

Quando i risultati di compute vengono restituiti, sono forniti come Series/DataFrame Pandas o array NumPy invece che come DataFrame Dask nativi.

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


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

pandas.core.series.Series

Il motivo è che la maggior parte delle operazioni di manipolazione dati restituisce solo un sottoinsieme del dataframe originale, occupando molto meno spazio. Quindi non ci sarà bisogno di usare il parallelismo di Dask e puoi continuare il resto del workflow in pandas o NumPy.

Dask Bag e Dask Delayed per dati non strutturati

Dask Bag e Dask Delayed sono due componenti della libreria Dask che offrono potenti strumenti per lavorare con dati non strutturati o semi-strutturati e per abilitare la valutazione pigra.

Se in passato i dati tabellari erano i più comuni, i dataset di oggi spesso includono file non strutturati come immagini, file di testo, video e audio. Dask Bag fornisce la funzionalità e l’API per gestire tali file non strutturati in modo parallelo e scalabile.

Ad esempio, consideriamo una semplice illustrazione:

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']

In questo esempio creiamo una Dask Bag b da una lista di stringhe. Applichiamo poi operazioni sulla Bag per filtrare le stringhe che iniziano con la lettera "a" e convertirle in maiuscolo usando rispettivamente le funzioni filter() e map(). Infine, calcoliamo il risultato come lista usando il metodo compute() e stampiamo l’output.

Ora immagina di poter eseguire operazioni ancora più complesse su miliardi di stringhe simili archiviate in un file di testo. Senza la valutazione pigra e il parallelismo offerti da Dask Bag, affronteresti sfide significative. (Per saperne di più su Bag, consulta la documentazione di Dask).

Per quanto riguarda Dask Delayed, offre ancora più flessibilità e introduce valutazione pigra e parallelismo in vari altri scenari. Con Dask Delayed, puoi convertire qualsiasi funzione Python nativa in un oggetto pigro usando il decorator @dask.delayed.

Ecco un semplice esempio:

%%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

In questo esempio definiamo una funzione process_data decorata con @dask.delayed. La funzione simula un po’ di lavoro computazionale dormendo per 1 secondo e poi restituendo il quadrato del valore di input.

Senza parallelismo, eseguire questo calcolo su 1000 input avrebbe richiesto più di 1000 secondi. Con Dask Delayed e l’esecuzione parallela, invece, il calcolo ha impiegato solo circa 42,1 secondi.

Questo esempio dimostra la potenza del parallelismo nel ridurre i tempi di calcolo distribuendo in modo efficiente il carico di lavoro tra più core o worker.

Questo è il senso del parallelismo.

Conclusioni e risorse aggiuntive

Dask è una delle librerie cardine nell’ecosistema dei dati. Estende le funzionalità di molte librerie amate come NumPy, Pandas e Spark, permettendo di gestire senza soluzione di continuità dataset più grandi della memoria.

Con Dask Bag e Dask Delayed, porta parallelismo e valutazione pigra in scenari non tradizionali, come il lavoro con dati non strutturati o con oggetti Python nativi.

Per i dettagli più tecnici, assicurati di leggere a fondo la documentazione di Dask.

Se cerchi una risorsa completa per padroneggiare Dask, dai un’occhiata al nostro corso  Parallel Programming With Dask in Python.