G1Joshi/Agent-Skills

Dask

Dask met à l'échelle Python. Il ressemble à Pandas/NumPy mais s'exécute sur des clusters. Les mises à jour 2025 se concentrent sur le High Performance Shuffle et l'intégration GPU.

Quand l'utiliser

• Calcul tabulaire dépassant la mémoire : Traitement de jeux de données CSV/Parquet de 10 Go à 1 To sur une machine unique ou un cluster multi-nœuds.
• Parallélisation de workflows Python personnalisés : Utilisation de dask.delayed pour paralléliser des boucles Python arbitraires et des graphes de tâches.
• Machine Learning distribué : Intégration avec Scikit-learn, XGBoost et LightGBM sur des nœuds workers distribués.
• Mise à l'échelle NumPy et Pandas : API familières prêtes à l'emploi pour les calculs sur des tableaux et DataFrames hors mémoire.

Démarrage rapide

import dask.dataframe as dd

# Lire paresseusement des fichiers CSV partitionnés de plusieurs Go
df = dd.read_csv('data/transactions_*.csv')

# Calculer des statistiques agrégées en parallèle sur tous les cœurs CPU
resultat = df.groupby('category').amount.mean().compute()
print(resultat)

Concepts de base

#Traitement hors mémoire avec Dask DataFrame

Chargement de fichiers Parquet partitionnés et calcul d'agrégations de manière paresseuse :

import dask.dataframe as dd
from dask.distributed import Client

# Initialiser un cluster distribué local
client = Client(n_workers=4, threads_per_worker=2, memory_limit='4GB')
print(f"Tableau de bord Dask accessible à : {client.dashboard_link}")

# Chargement paresseux d'un ensemble de fichiers
import dask.dataframe as dd
df = dd.read_parquet(
    's3://analytics-bucket/transactions/*.parquet',
    columns=['customer_id', 'amount', 'status', 'country'],
    engine='pyarrow'
)

# Construire le graphe de calcul paresseux
filtre = df[df['status'] == 'completed']
metriques = filtre.groupby('country')['amount'].agg(['mean', 'sum', 'count'])

# Exécuter le graphe et retourner un DataFrame Pandas concret
resultat_df = metriques.compute()
print(resultat_df)

#Graphes de tâches personnalisés avec dask.delayed

Parallélisation d'exécutions de fonctions indépendantes :

from dask import delayed, compute
import time

@delayed
def fetch_api_data(endpoint: str) -> dict:
    time.sleep(0.5) # Simulation d'E/S
    return {'endpoint': endpoint, 'status': 200, 'items': [1, 2, 3]}

@delayed
def transform_payload(data: dict) -> int:
    return sum(data['items'])

@delayed
def aggregate_totals(totals: list[int]) -> int:
    return sum(totals)

# Construire le DAG de dépendances
endpoints = ['users', 'orders', 'inventory', 'payments']
recuperes = [fetch_api_data(ep) for ep in endpoints]
transformes = [transform_payload(f) for f in recuperes]
total_final = aggregate_totals(transformes)

# Exécuter toutes les tâches en parallèle sur le cluster
total_resultat = total_final.compute()
print("Grand Total :", total_resultat)

#Dask Array pour l'algèbre linéaire distribuée

Manipulation de tableaux multi-dimensionnels massifs :

import dask.array as da

# Créer un tableau de 50 000 x 50 000 divisé en morceaux de 5 000 x 5 000
x = da.random.normal(10, 0.1, size=(50000, 50000), chunks=(5000, 5000))

# Effectuer des calculs paresseux
y = x + x.T
z = y[::2, ::2].mean(axis=0)

# Calculer le résultat
resultat = z.compute()
print("Forme moyenne du tableau calculé :", resultat.shape)

Patterns courants

Futures distribués pour des graphes de tâches irréguliers

Problème : Calculs parallèles personnalisés qui ne s'adaptent pas facilement aux structures lignes/colonnes de DataFrame/Array.

Solution : Utiliser le client distribué Dask avec des futures asynchrones :

from dask.distributed import Client

client = Client(n_workers=4, threads_per_worker=2)

def entrainer_partition(id_part : int):
    # Entraîner un sous-modèle local
    return f"Modèle {id_part} entraîné"

futures = [client.submit(entrainer_partition, i) for i in range(10)]
resultats = client.gather(futures)
print(resultats)

Meilleures pratiques (2026)

• À faire : Vérifier le tableau de bord web Dask (généralement sur le port 8787) pour surveiller la pression mémoire, les flux de tâches et les goulots d'étranglement.
• À faire : Choisir des tailles de morceaux entre 100 Mo et 300 Mo en mémoire pour une efficacité parallèle optimale.
• À faire : Persister les DataFrames intermédiaires (df = df.persist()) lors de l'interrogation répétée des mêmes données transformées.
• À faire : Filtrer les colonnes et les lignes tôt en utilisant des projections de colonnes et des prédicats de pushdown.
• À ne pas faire : Appeler .compute() dans des boucles ; construire le graphe de tâches complet et appeler compute() une seule fois.
• À ne pas faire : Utiliser Dask si les données tiennent confortablement en RAM ; Pandas natif et Polars sont nettement plus rapides pour les tâches en mémoire.
• À ne pas faire : Créer des millions de minuscules tâches delayed ; une surcharge excessive de tâches dégrade les performances du planificateur.

Dépannage

Erreur	Cause	Solution
KilledWorker: Worker exceeded 95% memory budget	Une partition de tâche est trop grande pour tenir dans la RAM du worker pendant le calcul.	Augmenter le nombre de partitions : df.repartition(npartitions=100).
UserWarning: Sending large object to workers	Un grand tableau NumPy ou DataFrame est passé comme argument direct de fonction.	Diffuser les données une fois avec client.scatter(data) avant de soumettre les tâches.
Slow compute: Too many tiny tasks	Des milliers de micro-partitions entraînent une surcharge de planification des tâches.	Fusionner les petites partitions avec repartition(partition_size="100MB").

Références

• Documentation Dask