dask

Que fait dask ?

Présentation

Dask est une bibliothèque Python pour le calcul parallèle et distribué qui offre trois capacités essentielles :

• Exécution plus volumineuse que la mémoire sur des machines uniques pour des données dépassant la RAM disponible
• Traitement parallèle pour améliorer la vitesse de calcul sur plusieurs cœurs
• Calcul distribué prenant en charge des datasets à l'échelle du téraoctet sur plusieurs machines

Dask passe à l'échelle des ordinateurs portables (traitement d'environ 100 Gio) aux clusters (traitement d'environ 100 Tio) tout en conservant les API Python familières.

Version en amont actuelle : dask 2026.3.0 (PyPI, mars 2026). Documentation : docs.dask.org. Depuis 2025.1.0, l'API DataFrame basée sur les expressions avec planification de requêtes est la seule implémentation — n'installez pas dask-expr séparément et ne définissez pas dataframe.query-planning: False.

Démarrage rapide

Installation

uv pip install "dask>=2025.1"

Pour un workflow pandas/NumPy typique avec le planificateur distribué et le tableau de bord :

uv pip install "dask[complete]"

Stockage d'objets distants (S3, GCS, Azure) :

uv pip install s3fs    # chemins s3://
uv pip install gcsfs   # chemins gs://

Nécessite Python 3.10+ (le support de Python 3.9 a été abandonné dans la version 2024.12). Les E/S DataFrame nécessitent PyArrow 16+ (depuis dask 2026.1.2).

Quand utiliser cette compétence

Cette compétence doit être utilisée lorsque :

• Vous traitez des datasets qui dépassent la RAM disponible
• Vous souhaitez passer à l'échelle des opérations pandas ou NumPy sur des datasets plus volumineux
• Vous parallélisez des calculs pour améliorer les performances
• Vous traitez plusieurs fichiers efficacement (CSV, Parquet, JSON, fichiers journaux)
• Vous construisez des workflows parallèles personnalisés avec des dépendances entre tâches
• Vous distribuez des charges de travail sur plusieurs cœurs ou machines

Capacités principales

Dask fournit cinq composants principaux, chacun adapté à des cas d'utilisation différents :

1. DataFrames - Opérations pandas parallèles

Objectif : Passer à l'échelle des opérations pandas sur des datasets plus volumineux grâce au traitement parallèle.

Quand les utiliser :

• Les données tabulaires dépassent la RAM disponible
• Vous devez traiter plusieurs fichiers CSV/Parquet ensemble
• Les opérations pandas sont lentes et nécessitent une parallélisation
• Passage à l'échelle d'un prototype pandas à la production

Documentation de référence : Pour des conseils complets sur les DataFrames Dask, reportez-vous à references/dataframes.md qui inclut :

• Lecture des données (fichiers uniques, fichiers multiples, motifs glob)
• Opérations courantes (filtrage, groupby, jointures, agrégations)
• Opérations personnalisées avec map_partitions
• Conseils d'optimisation des performances
• Modèles courants (ETL, séries temporelles, traitement multi-fichiers)

Exemple rapide :

import dask.dataframe as dd

# Lire plusieurs fichiers comme un seul DataFrame
ddf = dd.read_csv('data/2024-*.csv')

# Les opérations sont paresseuses jusqu'à compute()
filtered = ddf[ddf['value'] > 100]
result = filtered.groupby('category').mean().compute()

Points clés :

• Les opérations sont paresseuses (construction d'un graphe de tâches) jusqu'à l'appel de .compute()
• Utilisez map_partitions pour des opérations personnalisées efficaces
• Convertissez tôt en DataFrame lorsque vous travaillez avec des données structurées provenant d'autres sources

2. Arrays - Opérations NumPy parallèles

Objectif : Étendre les capacités de NumPy à des datasets plus volumineux que la mémoire en utilisant des algorithmes par blocs.

Quand les utiliser :

• Les tableaux dépassent la RAM disponible
• Les opérations NumPy nécessitent une parallélisation
• Vous travaillez avec des datasets scientifiques (HDF5, Zarr, NetCDF)
• Vous avez besoin d'algèbre linéaire ou d'opérations sur des tableaux en parallèle

Documentation de référence : Pour des conseils complets sur les Arrays Dask, reportez-vous à references/arrays.md qui inclut :

• Création de tableaux (à partir de NumPy, aléatoires, à partir du disque)
• Stratégies de découpage en chunks et optimisation
• Opérations courantes (arithmétique, réductions, algèbre linéaire)
• Opérations personnalisées avec map_blocks
• Intégration avec HDF5, Zarr et XArray

Exemple rapide :

import dask.array as da

# Créer un grand tableau avec des chunks
x = da.random.random((100000, 100000), chunks=(10000, 10000))

# Les opérations sont paresseuses
y = x + 100
z = y.mean(axis=0)

# Calculer le résultat
result = z.compute()

Points clés :

• La taille des chunks est cruciale (visez environ 100 Mo par chunk)
• Les opérations travaillent sur les chunks en parallèle
• Réorganisez les chunks (rechunk) si nécessaire pour des opérations efficaces
• Utilisez map_blocks pour des opérations non disponibles dans Dask

3. Bags - Traitement parallèle de données non structurées

Objectif : Traiter des données non structurées ou semi-structurées (texte, JSON, journaux) avec des opérations fonctionnelles.

Quand les utiliser :

• Traitement de fichiers texte, journaux ou enregistrements JSON
• Nettoyage des données et ETL avant une analyse structurée
• Travail avec des objets Python qui ne correspondent pas aux formats tableau/DataFrame
• Besoin d'un traitement en streaming économe en mémoire

Documentation de référence : Pour des conseils complets sur les Bags Dask, reportez-vous à references/bags.md qui inclut :

• Lecture de fichiers texte et JSON
• Opérations fonctionnelles (map, filter, fold, groupby)
• Conversion en DataFrames
• Modèles courants (analyse de journaux, traitement JSON, traitement de texte)
• Considérations de performance

Exemple rapide :

import dask.bag as db
import json

# Lire et analyser les fichiers JSON
bag = db.read_text('logs/*.json').map(json.loads)

# Filtrer et transformer
valid = bag.filter(lambda x: x['status'] == 'valid')
processed = valid.map(lambda x: {'id': x['id'], 'value': x['value']})

# Convertir en DataFrame pour l'analyse
ddf = processed.to_dataframe()

Points clés :

• Utilisez-les pour le nettoyage initial des données, puis convertissez en DataFrame/Array
• Utilisez foldby au lieu de groupby pour de meilleures performances
• Les opérations sont en streaming et économes en mémoire
• Convertissez dans des formats structurés (DataFrame) pour les opérations complexes

4. Futures - Parallélisation basée sur les tâches

Objectif : Construire des workflows parallèles personnalisés avec un contrôle fin de l'exécution des tâches et de leurs dépendances.

Quand les utiliser :

• Construction de workflows dynamiques et évolutifs
• Besoin d'exécution immédiate des tâches (pas de paresse)
• Les calculs dépendent de conditions d'exécution
• Implémentation d'algorithmes parallèles personnalisés
• Besoin de calculs avec état

Documentation de référence : Pour des conseils complets sur les Futures Dask, reportez-vous à references/futures.md qui inclut :

• Configuration du client distribué
• Soumission de tâches et travail avec les futures
• Dépendances entre tâches et déplacement des données
• Coordination avancée (files d'attente, verrous, événements, acteurs)
• Modèles courants (exploration de paramètres, tâches dynamiques, algorithmes itératifs)

Exemple rapide :

from dask.distributed import Client

client = Client()  # Créer un cluster local

# Soumettre les tâches (exécution immédiate)
def process(x):
    return x ** 2

futures = client.map(process, range(100))

# Récupérer les résultats
results = client.gather(futures)

client.close()

Points clés :

• Nécessite un client distribué (même pour une seule machine)
• Les tâches s'exécutent immédiatement lors de la soumission
• Pré-diffusez les grandes données pour éviter des transferts répétés
• Environ 1 ms de surcharge par tâche (pas adapté à des millions de petites tâches)
• Utilisez les acteurs pour les workflows avec état

5. Planificateurs (Schedulers) - Backends d'exécution

Objectif : Contrôler comment et où les tâches Dask s'exécutent (threads, processus, distribué).

Quand choisir un planificateur :

• Threads (par défaut) : Opérations NumPy/Pandas, bibliothèques libérant le GIL, avantage de la mémoire partagée
• Processus : Code Python pur, traitement de texte, opérations liées au GIL
• Synchrone : Débogage avec pdb, profilage, compréhension des erreurs
• Distribué : Besoin du tableau de bord, clusters multi-machines, fonctionnalités avancées

Documentation de référence : Pour des conseils complets sur les planificateurs Dask, reportez-vous à references/schedulers.md qui inclut :

• Descriptions détaillées et caractéristiques des planificateurs
• Méthodes de configuration (globale, gestionnaire de contexte, par calcul)
• Considérations de performance et surcharge
• Modèles courants et dépannage
• Configuration des threads pour des performances optimales

Exemple rapide :

import dask
import dask.dataframe as dd

# Utiliser les threads pour DataFrame (par défaut, bon pour le numérique)
ddf = dd.read_csv('data.csv')
result1 = ddf.mean().compute()  # Utilise les threads

# Utiliser les processus pour les tâches lourdes en Python
import dask.bag as db
bag = db.read_text('logs/*.txt')
result2 = bag.map(python_function).compute(scheduler='processes')

# Utiliser le mode synchrone pour le débogage
dask.config.set(scheduler='synchronous')
result3 = problematic_computation.compute()  # Peut utiliser pdb

# Utiliser le mode distribué pour le suivi et la mise à l'échelle
from dask.distributed import Client
client = Client()
result4 = computation.compute()  # Utilise le mode distribué avec tableau de bord

Points clés :

• Threads : Surcharge la plus faible (~10 µs/tâche), idéal pour les tâches numériques
• Processus : Évite le GIL (~10 ms/tâche), idéal pour les tâches Python
• Distribué : Tableau de bord de suivi (~1 ms/tâche), passe à l'échelle sur les clusters
• Vous pouvez changer de planificateur par calcul ou globalement

Bonnes pratiques

Pour des conseils complets sur l'optimisation des performances, les stratégies de gestion de la mémoire et les pièges courants à éviter, reportez-vous à references/best-practices.md. Les principes clés incluent :

Commencez par des solutions plus simples

Avant d'utiliser Dask, explorez :

• De meilleurs algorithmes
• Des formats de fichiers efficaces (Parquet plutôt que CSV)
• Du code compilé (Numba, Cython)
• L'échantillonnage des données

Règles de performance critiques

1. Ne chargez pas les données localement avant de les confier à Dask

# Erreur : Charge toutes les données en mémoire d'abord
import pandas as pd
df = pd.read_csv('large.csv')
ddf = dd.from_pandas(df, npartitions=10)

# Correct : Laissez Dask gérer le chargement
import dask.dataframe as dd
ddf = dd.read_csv('large.csv')

2. Évitez les appels répétés à compute()

# Erreur : Chaque compute est séparé
for item in items:
    result = dask_computation(item).compute()

# Correct : Un seul compute pour tout
computations = [dask_computation(item) for item in items]
results = dask.compute(*computations)

3. Ne construisez pas des graphes de tâches excessivement grands

• Augmentez la taille des chunks s'il y a des millions de tâches
• Utilisez map_partitions/map_blocks pour fusionner les opérations
• Vérifiez la taille du graphe de tâches : len(ddf.__dask_graph__())

4. Choisissez des tailles de chunks appropriées

• Cible : ~100 Mo par chunk (ou 10 chunks par cœur dans la mémoire du worker)
• Trop grand : Dépassement mémoire
• Trop petit : Surcharge de planification

5. Utilisez le tableau de bord

from dask.distributed import Client
client = Client()
print(client.dashboard_link)  # Surveillez les performances, identifiez les goulots d'étranglement

Modèles de workflow courants

Pipeline ETL

import dask.dataframe as dd

# Extract : Lire les données
ddf = dd.read_csv('raw_data/*.csv')

# Transform : Nettoyer et traiter
ddf = ddf[ddf['status'] == 'valid']
ddf['amount'] = ddf['amount'].astype('float64')
ddf = ddf.dropna(subset=['important_col'])

# Load : Agréger et sauvegarder
summary = ddf.groupby('category').agg({'amount': ['sum', 'mean']})
summary.to_parquet('output/summary.parquet')

Pipeline non structuré à structuré

import dask.bag as db
import json

# Commencez par un Bag pour les données non structurées
bag = db.read_text('logs/*.json').map(json.loads)
bag = bag.filter(lambda x: x['status'] == 'valid')

# Convertissez en DataFrame pour une analyse structurée
ddf = bag.to_dataframe()
result = ddf.groupby('category').mean().compute()

Calcul de tableau à grande échelle

import dask.array as da

# Charger ou créer un grand tableau
x = da.from_zarr('large_dataset.zarr')

# Traiter par chunks
normalized = (x - x.mean()) / x.std()

# Sauvegarder le résultat (utilisez mode= pour l'écrasement ; zarr_array_kwargs pour la compression)
da.to_zarr(normalized, 'normalized.zarr', mode='w')

Workflow parallèle personnalisé

from dask.distributed import Client

client = Client()

# Diffusez le grand dataset une fois
data = client.scatter(large_dataset)

# Traitez en parallèle avec des dépendances
futures = []
for param in parameters:
    future = client.submit(process, data, param)
    futures.append(future)

# Récupérez les résultats
results = client.gather(futures)

Choisir le bon composant

Utilisez ce guide de décision pour choisir le composant Dask approprié :

Type de données :

• Données tabulaires → DataFrames
• Tableaux numériques → Arrays
• Texte/JSON/journaux → Bags (puis convertir en DataFrame)
• Objets Python personnalisés → Bags ou Futures

Type d'opération :

• Opérations pandas standard → DataFrames
• Opérations NumPy standard → Arrays
• Tâches parallèles personnalisées → Futures
• Traitement de texte/ETL → Bags

Niveau de contrôle :

• Haut niveau, automatique → DataFrames/Arrays
• Bas niveau, manuel → Futures

Type de workflow :

• Graphe de calcul statique → DataFrames/Arrays/Bags
• Dynamique, évolutif → Futures

Considérations d'intégration

Formats de fichiers

• Efficaces : Parquet, HDF5, Zarr (colonne, compressé, favorable au parallélisme)
• Compatibles mais plus lents : CSV (à utiliser uniquement pour l'ingestion initiale)
• Pour les Arrays : HDF5, Zarr, NetCDF

Conversion entre collections

# Bag → DataFrame
ddf = bag.to_dataframe()

# DataFrame → Array (pour les données numériques)
arr = ddf.to_dask_array(lengths=True)

# Array → DataFrame
ddf = dd.from_dask_array(arr, columns=['col1', 'col2'])

Avec d'autres bibliothèques

• XArray : Enveloppe les tableaux Dask avec des dimensions étiquetées (géospatial, imagerie)
• Dask-ML : Apprentissage automatique avec des API compatibles scikit-learn
• Distributed : Gestion avancée des clusters et surveillance

Débogage et développement

Workflow de développement itératif

1. Testez sur de petites données avec le planificateur synchrone :

dask.config.set(scheduler='synchronous')
result = computation.compute()  # Peut utiliser pdb, débogage facile

2. Validez avec des threads sur un échantillon :

sample = ddf.head(1000)  # Petit échantillon
# Testez la logique, puis passez à l'échelle sur le dataset complet

3. Passez à l'échelle avec le mode distribué pour la surveillance :

from dask.distributed import Client
client = Client()
print(client.dashboard_link)  # Surveillez les performances
result = computation.compute()

Problèmes courants

Erreurs mémoire :

• Diminuez la taille des chunks
• Utilisez persist() stratégiquement et supprimez après usage
• Vérifiez les fuites mémoire dans les fonctions personnalisées

Démarrage lent :

• Graphe de tâches trop grand (augmentez la taille des chunks)
• Utilisez map_partitions ou map_blocks pour réduire le nombre de tâches

Parallélisation médiocre :

• Chunks trop grands (augmentez le nombre de partitions)
• Utilisation de threads avec du code Python (passez aux processus)
• Dépendances de données empêchant le parallélisme (utilisez repartition ou ajustez les clés)

Problèmes de planification :

• Trop de tâches (augmentez la taille des chunks)
• Fonctions trop rapides (regroupez les tâches)
• Surcharge du planificateur (utilisez le mode distribué)

Références

Pour des références détaillées sur chaque composant, consultez les fichiers suivants :

• references/dataframes.md - Guide complet sur les DataFrames
• references/arrays.md - Guide complet sur les Arrays
• references/bags.md - Guide complet sur les Bags
• references/futures.md - Guide complet sur les Futures
• references/schedulers.md - Guide complet sur les planificateurs
• references/best-practices.md - Optimisation des performances et bonnes pratiques