Introduzione ai DataFrame Dask

Dask è una libreria di calcolo parallelo per Python che estende NumPy e Pandas a dataset più grandi della RAM. Il modulo dask.dataframe fornisce un'API DataFrame quasi identica a quella di Pandas, ma invece di eseguire immediatamente le operazioni, Dask costruisce un grafo di attività e le esegue in modo differito quando si chiama .compute(). In questo modo Dask può parallelizzare il lavoro su più core o persino su più macchine, con modifiche minime al codice.

Dask si installa con pip install dask[dataframe]. La convenzione per l'importazione è import dask.dataframe as dd. Internamente, un DataFrame Dask viene suddiviso in molti DataFrame Pandas più piccoli, ciascuno elaborato in modo indipendente. Le operazioni sul DataFrame Dask creano un grafo di attività differite: nulla viene eseguito finché non si chiama .compute(). Questa separazione tra la descrizione e l'esecuzione del calcolo è l'idea chiave di Dask.

Con Pandas, ogni operazione viene eseguita immediatamente e in modo eager. Con Dask, le operazioni restituiscono un altro oggetto Dask che rappresenta il calcolo differito. Solo quando si chiama .compute() Dask legge effettivamente i dati ed esegue il grafo di attività. Questa esecuzione differita consente a Dask di ottimizzare il piano prima dell'esecuzione: per esempio, può fondere filtri consecutivi per evitare di caricare i dati più volte. È come una ricetta: Dask scrive la ricetta, mentre .compute() prepara il piatto.

Un DataFrame Dask è suddiviso in partizioni, ognuna delle quali è un normale DataFrame Pandas. Per impostazione predefinita, dd.read_csv crea una partizione per file (oppure una partizione ogni 128 MB per i file di grandi dimensioni). È possibile controllare questo comportamento con blocksize. Verificando ddf.npartitions si visualizza il numero di partizioni esistenti. Un numero maggiore di partizioni aumenta il parallelismo, ma aggiunge overhead; un numero minore riduce l'overhead, ma limita il parallelismo. Il compromesso ideale consiste in genere in alcune centinaia di partizioni.

La maggior parte delle operazioni comuni di Pandas funziona allo stesso modo in Dask: .head(), .tail(), .describe(), l'indicizzazione booleana, .groupby(), .merge() e .assign() hanno tutte equivalenti in Dask. La differenza principale è che è necessario chiamare .compute() per materializzare il risultato. Le operazioni che Pandas gestisce in millisecondi possono richiedere secondi in Dask a causa dell'overhead del grafo di attività: utilizzi quindi Pandas per i dati di piccole dimensioni e Dask quando i dati non entrano nella RAM.

Una delle funzionalità più utili di Dask è la lettura di più file contemporaneamente tramite pattern glob. dd.read_csv('data/2024-*.csv') legge tutti i file corrispondenti e crea una partizione per file. È ideale per i dati archiviati in file suddivisi per mese o per giorno, uno schema comune nei data lake. Dask allinea automaticamente gli schemi: è l'equivalente di un ciclo manuale con concatenazione in Pandas, ma molto più semplice.

Prima di eseguire una pipeline Dask complessa, è possibile esaminare il grafo di attività chiamando result.visualize(), che genera un diagramma PNG di tutti i passaggi del calcolo. È utile per capire cosa eseguirà Dask e per individuare le cause di rallentamenti imprevisti. Il grafo mostra come le partizioni passano attraverso i passaggi di filtraggio, raggruppamento e aggregazione, rendendo facile individuare i calcoli ridondanti. È richiesto il pacchetto graphviz.

Quando è necessario applicare una funzione Pandas personalizzata a un DataFrame Dask, si utilizzi ddf.map_partitions(func). Questo applica func a ogni partizione in modo indipendente e restituisce un nuovo DataFrame Dask. La funzione riceve un normale DataFrame Pandas e deve restituirne uno. È l'equivalente Dask di df.apply() e consente di integrare Dask con codice che comprende soltanto Pandas.

Dask dispone di più scheduler che controllano il modo in cui vengono eseguite le attività. Lo scheduler 'synchronous' esegue le attività in sequenza nel thread corrente (utile per il debugging). Lo scheduler 'threads' utilizza un pool di thread (adatto alle operazioni vincolate dall'I/O). Lo scheduler 'processes' avvia più processi per le operazioni vincolate dalla CPU (aggirando il GIL di Python). Un cluster Dask distribuito consente l'esecuzione su più macchine. Specifichi lo scheduler tramite compute(scheduler='threads').

È comune elaborare un dataset di grandi dimensioni con Dask e poi trasferire il risultato aggregato in Pandas per l'analisi o la visualizzazione finale. Utilizzi .compute() per convertire un DataFrame Dask in Pandas. Nella direzione opposta, dd.from_pandas(df, npartitions=4) converte un DataFrame Pandas in un DataFrame Dask, utile per testare il codice Dask su dati di piccole dimensioni prima di estenderlo all'intero dataset.

Dask non è sempre lo strumento più adatto. Utilizzi Pandas quando i dati entrano nella RAM (meno di qualche GB): è più semplice e veloce grazie al minore overhead. Utilizzi Dask quando i dati superano la RAM, ma desidera una sintassi simile a Pandas e il parallelismo su una singola macchina. Utilizzi SQL o un database quando i dati risiedono in un database relazionale e le aggregazioni possono essere delegate al motore del database. Utilizzi Spark o BigQuery quando è necessaria un'elaborazione distribuita su più macchine, su scala petabyte.

Verifichi la propria comprensione dei concetti di analisi dei dati trattati in questa lezione.

In questa lezione ha imparato che i DataFrame Dask sono raccolte di partizioni Pandas valutate in modo differito e dotate di un'API familiare, che .compute() avvia l'esecuzione effettiva del grafo di attività e che map_partitions consente di applicare qualsiasi funzione Pandas personalizzata a tutte le partizioni. Prossimamente esamineremo il formato Parquet, un'alternativa colonnare e veloce al CSV per l'archiviazione di dataset di grandi dimensioni.