Spark & PySpark

Que fait Spark & PySpark ?

Presque tous les jobs Spark lents le sont pour l'une des trois raisons suivantes : un shuffle évitable, une clé asymétrique (skew), ou des lignes qui transitent par Python une par une. L'erreur coûteuse est de dimensionner le cluster avant de lire le plan d'exécution — payer pour plus d'exécuteurs pour exécuter le même DAG inefficace plus rapidement.

Procédure opératoire

Étape 1 : Rassembler les informations

1. Le code du job et, pour un job lent, la vue Spark UI de l'étape lente (distribution des durées des tâches, tailles de shuffle en lecture/écriture) et la sortie de df.explain(True).
2. La taille et le format des données en entrée, ainsi que les tailles des deux côtés de chaque jointure.
3. La forme du cluster : nombre d'exécuteurs, cœurs et mémoire.
4. Si le job est une nouvelle rédaction ou une passe d'optimisation - pour l'optimisation, diagnostiquer à partir du plan avant de toucher au code.

Étape 2 : Démarrer avec une session saine

from pyspark.sql import SparkSession
from pyspark.sql import functions as F

spark = (
    SparkSession.builder
    .appName("etl")
    .config("spark.sql.shuffle.partitions", "200")
    .config("spark.sql.adaptive.enabled", "true")
    .getOrCreate()
)

Activez l'exécution de requêtes adaptative (AQE) ; elle fusionne les partitions de shuffle et change les stratégies de jointure à l'exécution. Ensuite, dimensionnez délibérément spark.sql.shuffle.partitions : ciblez 100-200 Mo par partition de shuffle, donc partitions ≈ taille des données de shuffle ÷ 128 Mo, et au moins 2-3 fois le nombre total de cœurs d'exécuteurs pour qu'aucun cœur ne reste inactif.

Étape 3 : Restez dans l'API DataFrame et évitez Python

Les opérations DataFrame passent par Catalyst et Tungsten, offrant l'optimisation des requêtes et la mémoire hors tas. Passez aux RDD uniquement lorsqu'aucune primitive DataFrame n'existe. Les UDF Python sérialisent chaque ligne vers le worker Python et retour, cassant la génération de code. Ordre de préférence :

1. Fonctions intégrées pyspark.sql.functions (ex. F.regexp_extract, F.when).
2. UDF Pandas (vectorisées) lorsque Python est inévitable.
3. UDF Python scalaires uniquement en dernier recours.

from pyspark.sql.functions import pandas_udf

@pandas_udf("double")
def normalize(s):
    return (s - s.mean()) / s.std()

Étape 4 : Choisir la stratégie de jointure par taille

• Diffusez (broadcast) le petit côté lorsqu'il tient en mémoire. Spark diffuse automatiquement en dessous de spark.sql.autoBroadcastJoinThreshold (10 Mo par défaut) ; diffusez explicitement des tables jusqu'à quelques centaines de Mo lorsque la mémoire de l'exécuteur le permet, et considérez ~1 Go comme le plafond pratique — la diffusion copie la table sur chaque exécuteur et provoque un OOM du job si mal évaluée.

from pyspark.sql.functions import broadcast
result = large.join(broadcast(small), "key")

• Filtrez et projetez les colonnes avant la jointure pour réduire le shuffle ; chaque colonne et ligne supprimée est un octet de shuffle économisé.
• Deux grandes tables → acceptez la jointure sort-merge, mais vérifiez d'abord le skew (Étape 5).

Étape 5 : Diagnostiquer et corriger le skew

Symptômes de skew : l'étape est terminée sauf une ou deux tâches ; la durée maximale des tâches dépasse ~3-5x la médiane ; une partition de lecture du shuffle est bien plus grande que les autres dans l'interface Spark UI. La gestion du skew de jointure d'AQE divise une partition lorsqu'elle est à la fois supérieure à 5x la taille médiane des partitions et supérieure à 256 Mo (les valeurs par défaut) — vérifiez qu'elle s'est déclenchée dans le plan. Lorsqu'AQE ne peut pas aider (par exemple, agrégation asymétrique), salez les clés chaudes : ajoutez un suffixe aléatoire 0..N à la clé du grand côté, multipliez le petit côté sur tous les N suffixes, joignez, puis retirez le sel. Choisissez N ≈ le facteur de skew (une clé 20x la médiane obtient N=20).

Étape 6 : Contrôler le partitionnement et la disposition des fichiers de sortie

• repartition(n, col) effectue un shuffle complet pour équilibrer les données ; utilisez-le avant les écritures larges.
• coalesce(n) réduit le nombre de partitions sans shuffle ; utilisez-le pour éviter les fichiers de sortie trop petits.
• Visez des fichiers de sortie de 128 Mo à 1 Go ; des milliers de petits fichiers pénalisent tous les lecteurs en aval.
• Partitionnez la sortie uniquement par des colonnes de faible cardinalité (date, région — des dizaines à centaines de valeurs, jamais des identifiants) :

df.write.partitionBy("dt").mode("overwrite").parquet("/data/out")

Étape 7 : Ne mettre en cache que ce qui est réutilisé

Ne mettez en cache que lorsqu'un DataFrame est utilisé plusieurs fois dans le DAG, matérialisez-le, et libérez-le après utilisation :

df.cache()
df.count()  # materialize
# ... reuse df ...
df.unpersist()

Mettre en cache un DataFrame utilisé une seule fois gaspille de la mémoire et peut évincer des données qui comptaient.

Étape 8 : Vérifier avec le plan

Réexécutez df.explain(True) et confirmez que la stratégie prévue apparaît (BroadcastHashJoin, filtres poussés, pas d'Exchange là où un a été éliminé). Lisez des formats columnaires (Parquet, ORC) pour que la prédication de filtres fonctionne, n'utilisez jamais collect() sur de grandes données vers le driver, et préférez les références F.col aux chaînes de caractères lors des chaînages.

Exemple pratique : mauvais vs bonne jointure

Mauvais - une table d'événements de 2 To jointe à une dimension de 200 Mo, toute la largeur, UDF Python pour l'analyse :

result = (events.join(dims, "store_id")          # sort-merge : shufle tous les 2 To
    .withColumn("region", parse_region_udf("meta"))  # Python ligne par ligne
    .filter(F.col("dt") == "2024-06-01"))            # filtre APRÈS la jointure

Le plan montre un Exchange des deux côtés (2 To shuffled), et l'UDF bloque la génération de code.

Bon - filtrez et projetez d'abord, diffusez la dimension, utilisez une fonction intégrée :

result = (events
    .filter(F.col("dt") == "2024-06-01")              # réduit à ~30 Go d'abord
    .select("store_id", "amount", "meta")
    .join(broadcast(dims.select("store_id", "region_code")), "store_id")
    .withColumn("region", F.regexp_extract("meta", r"region=(\w+)", 1)))

Le plan montre maintenant BroadcastHashJoin sans Exchange côté grand : ~30 Go scannés, zéro shuffle de la grande table, sans frontière Python.

Livrable

Produisez le code de job révisé ainsi qu'une note d'optimisation indiquant : la stratégie de jointure pour chaque jointure avec la preuve de taille, le paramètre de partitionnement de shuffle avec son calcul, tout skew trouvé (ratio max vs médiane des tâches) et la correction appliquée, les décisions de cache, les attentes de taille des fichiers de sortie, et les différences de explain avant/après pour une passe d'optimisation.

À NE PAS FAIRE

• Ne diffusez PAS une table dont vous n'avez pas mesuré la taille - une diffusion mal évaluée provoque un OOM sur tous les exécuteurs à la fois.
• N'écrivez PAS d'UDF Python avant de vérifier pyspark.sql.functions ; presque toute la logique sur les chaînes, les dates et les conditions a une fonction intégrée.
• N'utilisez PAS collect() ou toPandas() sur un grand DataFrame ; cela achemine les données du cluster à travers un seul driver.
• Ne partitionnez PAS la sortie par une colonne de haute cardinalité - des millions de répertoires, une ligne chacun.
• N'ajoutez PAS d'exécuteurs à un job dont l'étape lente est une seule tâche asymétrique ; plus de cœurs ne peuvent pas diviser une seule partition.
• Ne mettez PAS en cache par réflexe ; ne mettez en cache que les DataFrames réutilisés en aval, et libérez-les.

Barre de qualité

Un job terminé est considéré réussi uniquement lorsque : chaque jointure nomme sa stratégie et la preuve de taille ; spark.sql.shuffle.partitions est dérivé de la taille des données, pas laissé à une valeur par habitude ; l'interface Spark UI ne montre aucune tâche au-dessus de ~3x la durée médiane dans les étapes de shuffle (ou le skew est expliqué et corrigé) ; aucune UDF Python scalaire ne survit là où une fonction intégrée existe ; les fichiers de sortie se situent dans la bande 128 Mo-1 Go ; et le explain final confirme le plan prévu.