Booster votre workflow data science avec Dask : guide complet

Quand Wes McKinney a commencé à écrire pandas, il s'était fixé une règle empirique : pour que pandas fonctionne de façon optimale, la RAM de la machine doit être 5 à 10 fois plus grande que le jeu de données traité. C'était facile à respecter vers 2010, mais nous sommes désormais en 2023.

Dès 2020, les jeux de données réels avaient déjà atteint des tailles capables de faire planter des ordinateurs portables et des machines du quotidien. Anticipant ce problème, une solution est sortie dès 2015.

Dask est une bibliothèque open source développée par les créateurs d'Anaconda pour relever les défis du calcul scalable et efficace sur des jeux de données dépassant la capacité mémoire d'une seule machine.

Ce tutoriel propose une introduction complète à Dask et à ses fonctionnalités clés, notamment les interfaces pour DataFrames, Arrays et Bags.

Installer Dask

Comme toute autre bibliothèque, vous pouvez installer Dask de trois manières : Conda, Pip et depuis les sources.

Comme il s'agit d'un article d'introduction à Dask, nous ne couvrirons pas la dernière méthode, destinée aux mainteneurs.

Si vous utilisez Anaconda, Dask est inclus dans l'installation par défaut (preuve de sa popularité). Si vous souhaitez le réinstaller ou le mettre à niveau, utilisez la commande install :

conda install dask

L'alternative via PIP est la suivante :

pip install "dask[complete]"

L'ajout de l'extension [complete] installe également les dépendances requises de Dask, évitant d'avoir à installer manuellement NumPy, pandas et Tornado.

Vous pouvez vérifier si l'installation a réussi en consultant la version de la bibliothèque :

import dask

dask.__version__

Sortie :

'2023.5.0'

L'essentiel de votre travail avec Dask portera sur trois interfaces : Dask DataFrames, Arrays et Bags. Importons-les ainsi que numpy et pandas pour la suite de l'article :

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

Concepts de base de Dask

À un niveau élevé, vous pouvez voir Dask comme une surcouche qui étend les capacités d'outils traditionnels comme pandas, NumPy et Spark pour traiter des jeux de données plus volumineux que la mémoire.

Face à de grands objets comme des tableaux (vecteurs) ou des matrices (dataframes) plus gros que la mémoire, Dask les découpe en morceaux, appelés partitions.

Par exemple, considérons un tableau de 12 nombres aléatoires en NumPy et 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

L'image ci-dessus montre que le tableau Dask contient quatre blocs puisque nous avons défini chunks à 3. Sous le capot, chaque bloc est lui-même un tableau NumPy.

Passons maintenant à un exemple bien plus grand. Nous allons créer deux tableaux 10k par 100k (1 milliard d'éléments) et effectuer une multiplication élément par élément dans les deux bibliothèques tout en mesurant les performances :

# 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

Le code ci-dessus illustre la multiplication élément par élément de deux grands tableaux avec NumPy et Dask. Comme le montre la sortie, Dask est environ 34 fois plus rapide que NumPy pour ce calcul. Les gains deviennent encore plus significatifs à mesure que la complexité et la taille des tableaux augmentent.

Dask applique une approche similaire de découpage et de distribution des blocs sur tous les cœurs disponibles de votre machine pour d'autres objets également.

Dask DataFrames

Pour profiter pleinement des avantages de Dask, il nous faut un grand jeu de données, idéalement supérieur à 1 Go. Toutefois, télécharger un tel fichier pour suivre le tutoriel n'est pas optimal. À la place, vous pouvez utiliser ce script, qui génère un jeu de données synthétique de 10 millions de lignes, 10 variables numériques et 10 variables catégorielles.

Assurez-vous que votre machine dispose d'au moins 12 Go de RAM pour exécuter le script.

Une fois le fichier large_dataset.csv présent dans votre espace de travail, vous pouvez le charger avec la fonction read_csv de l'interface Dask DataFrames (dd) :

import dask.dataframe as dd

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

dask_df.head()

Même si le fichier est volumineux, vous remarquerez que le résultat s'affiche quasiment instantanément. Pour des fichiers encore plus grands, vous pouvez spécifier le paramètre blocksize, qui détermine le nombre d'octets utilisés pour découper le fichier.

Tout comme les Dask Arrays contiennent des blocs de petits tableaux NumPy, Dask est conçu pour gérer plusieurs petits DataFrames pandas juxtaposés le long de l'index de lignes.

Comme vous pouvez le deviner à partir de la fonction read_csv, la plupart de la syntaxe et des fonctionnalités courantes de l'API pandas sont conservées dans Dask. Les blocs de code suivants devraient vous être familiers si vous avez déjà travaillé avec pandas.

Sélection de colonnes et opérations élément par élément

Dans cet exemple, nous effectuons des opérations simples sur des colonnes de notre Dask DataFrame, appelé dask_df. Nous additionnons les valeurs de la colonne Numeric_0 au résultat de la multiplication des valeurs de Numeric_9 et Numeric_3. Nous stockons le résultat dans une variable nommée 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

Comme mentionné, Dask diffère des outils de calcul traditionnels : il n'exécute pas immédiatement ces opérations. Il crée plutôt un « plan » appelé graphe de tâches pour les réaliser plus tard. Cette approche permet à Dask d'optimiser et de paralléliser les calculs si besoin. La fonction compute() déclenche finalement l'exécution, et head() affiche les premières lignes du résultat.

Filtrage conditionnel

Voyons maintenant comment filtrer des données avec Dask. Nous sélectionnons les lignes dont la valeur de la colonne "Categorical_5" est "A".

Ce filtrage ressemble à ce que vous feriez avec pandas, mais avec une différence : Dask l'exécute de manière paresseuse. Il prépare le graphe de tâches, puis attend l'appel à compute() pour exécuter. En lançant head(), nous voyons les premières lignes du DataFrame filtré.

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

Statistiques descriptives courantes

Ensuite, nous allons produire des statistiques descriptives classiques avec la fonction describe() de Dask.

Elle fournit plusieurs indicateurs pour notre DataFrame, dont la moyenne, l'écart type, le minimum, le maximum, etc. Comme précédemment, Dask prépare le graphe de tâches à l'appel de describe(), mais n'exécute qu'au moment de 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

Nous utilisons également value_counts() pour compter le nombre d'occurrences de chaque valeur unique dans la colonne "Categorical_3". Nous déclenchons l'opération avec compute(), et head() affiche les valeurs les plus fréquentes.

Groupby

Enfin, utilisons la fonction groupby() pour regrouper les données selon les valeurs de la colonne "Categorical_8". Puis nous sélectionnons la colonne "Numeric_7" et calculons la moyenne pour chaque groupe.

C'est similaire à l'usage de groupby() dans pandas, mais, comme vous l'avez deviné, Dask l'exécute paresseusement. Nous déclenchons l'opération avec compute(), et head() affiche la moyenne de la colonne "Numeric_7" pour les premiers groupes.

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

Consultez cette section du guide utilisateur Dask pour le reste des similitudes entre pandas et Dask.

Évaluation paresseuse

Voyons maintenant l'usage de la fonction compute à la fin de chaque bloc de code.

Dask évalue les blocs de code en mode paresseux, contrairement au mode « eager » de pandas, qui renvoie les résultats immédiatement.

Pour filer une métaphore culinaire, l'évaluation paresseuse revient à préparer et émincer les ingrédients à l'avance, mais à ne les assembler pour cuire que lorsqu'on en a besoin. La fonction compute joue ce rôle.

À l'inverse, l'évaluation « eager » consiste à jeter les ingrédients au feu dès qu'ils sont prêts, afin que tout soit servi d'un seul coup.

L'évaluation paresseuse est essentielle aux excellentes performances de Dask, car elle permet :

1. Moins de calculs. Les expressions ne sont évaluées que lorsqu'elles sont nécessaires (à l'appel de compute), évitant des résultats intermédiaires inutiles.
2. Une allocation optimale des ressources. Elle évite d'allouer mémoire ou puissance de calcul à des résultats intermédiaires non requis.
3. Le support de grands jeux de données. Cette méthode traite les éléments à la volée ou par petits blocs, optimisant l'utilisation de la mémoire.

Lorsque les résultats de compute sont renvoyés, ils le sont sous forme de Series/DataFrames pandas ou de tableaux NumPy, et non de DataFrames Dask natifs.

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


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

pandas.core.series.Series

La raison est que la plupart des opérations de manipulation de données ne renvoient qu'un sous-ensemble du dataframe d'origine, beaucoup plus petit. Il n'est donc plus nécessaire d'utiliser le parallélisme de Dask, et vous pouvez poursuivre votre workflow dans pandas ou NumPy.

Dask Bags et Dask Delayed pour les données non structurées

Dask Bags et Dask Delayed sont deux composants de la bibliothèque Dask qui offrent des outils puissants pour travailler avec des données non structurées ou semi-structurées, tout en permettant l'évaluation paresseuse.

Autrefois, les données tabulaires dominaient, mais les jeux de données actuels contiennent souvent des fichiers non structurés comme des images, des fichiers texte, des vidéos et de l'audio. Dask Bags fournit les fonctionnalités et l'API pour traiter ces fichiers non structurés de manière parallèle et scalable.

Par exemple, voici une simple illustration :

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

Dans cet exemple, nous créons un Dask Bag b à partir d'une liste de chaînes. Nous filtrons ensuite les chaînes commençant par la lettre « a » et les convertissons en majuscules à l'aide des fonctions filter() et map(). Enfin, nous calculons le résultat sous forme de liste avec la méthode compute() et affichons la sortie.

Imaginez maintenant réaliser des opérations bien plus complexes sur des milliards de chaînes similaires stockées dans un fichier texte. Sans l'évaluation paresseuse et le parallélisme offerts par Dask Bags, la tâche serait ardue. (Plus d'informations sur Bags dans la documentation Dask.)

Concernant Dask Delayed, il offre encore plus de flexibilité et apporte l'évaluation paresseuse et le parallélisme à de nombreux autres scénarios. Avec Dask Delayed, vous pouvez convertir n'importe quelle fonction Python native en objet paresseux via le décorateur @dask.delayed.

Voici un exemple simple :

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

Dans cet exemple, nous définissons une fonction process_data décorée avec @dask.delayed. La fonction simule un travail de calcul en dormant 1 seconde, puis renvoie le carré de la valeur d'entrée.

Sans parallélisme, exécuter ce calcul sur 1000 entrées aurait pris plus de 1000 secondes. Avec Dask Delayed et l'exécution en parallèle, le calcul n'a pris qu'environ 42,1 secondes.

Cet exemple illustre la puissance du parallélisme pour réduire le temps de calcul en répartissant efficacement la charge sur plusieurs cœurs ou workers.

C'est tout l'intérêt du parallélisme.

Conclusion et ressources complémentaires

Dask est l'une des bibliothèques piliers de l'écosystème data. Il étend les fonctionnalités des bibliothèques plébiscitées comme NumPy, pandas et Spark, et permet de traiter sans couture des jeux de données plus volumineux que la mémoire.

Avec Dask Bags et Dask Delayed, il apporte parallélisme et évaluation paresseuse à des scénarios moins classiques, comme la manipulation de données non structurées ou d'objets Python natifs.

Pour entrer dans le détail, prenez le temps de parcourir la documentation Dask.

Si vous cherchez une ressource complète pour maîtriser Dask, découvrez notre cours  Parallel Programming With Dask in Python.