Optimisation des performances d'Apache Spark : guide pratique avec exemples

Introduction

Apache Spark traite rapidement de gros volumes de données quand on l'oriente pour bien utiliser les ressources du cluster. Ce guide montre comment identifier les goulots d'étranglement, dimensionner les exécuteurs et les partitions, optimiser les jointures et les mélanges (shuffles), et vérifier les améliorations avec des contrôles concrets dans Spark UI et les journaux. Vous obtiendrez un workflow sûr et incrémental avec des exemples pratiques pour optimiser des jobs sensibles à la latence ou orientés débit sans réécritures risquées.

Ce que vous apprendrez :

• Comment inventorier versions et topologie d'environnement pour que votre optimisation s'applique à votre réalité.
• Un chemin sûr et cadré pour ajuster le partitionnement, les stratégies de jointure, les shuffles, la mémoire, la sérialisation et le dimensionnement des ressources.
• Comment vérifier les améliorations avec Spark UI, SQL EXPLAIN, les journaux d'événements et des métriques simples de latence/débit.
• Les modes de défaillance courants (OOM, déséquilibre, déversements, queues longues) et comment revenir en arrière rapidement.
• Une liste de contrôle pratique pour maintenir les jobs en santé semaine après semaine.

Note sur les exemples : tous les chiffres dans les exemples sont construits pour illustrer l'approche, pas pour revendiquer des benchmarks spécifiques.

Inventaire des versions et de l'environnement

Avant de changer quoi que ce soit, figez le contexte. Cela évite de longs détours causés par des versions mal assorties ou des limites de ressources cachées.

Liste de contrôle d'inventaire :

• Version et build de Spark : distribution (Apache, fournisseur), version Scala/Python.
• JVM : version et mode GC (ex. G1).
• Gestionnaire de cluster : standalone, YARN ou Mesos ; statut de l'allocation dynamique.
• Nœuds et disques : nombre de cœurs, RAM, type et bande passante des disques locaux ; bande passante réseau.
• Stockage : HDFS, objet (ex. compatible S3) ou local ; formats de fichiers et compression par défaut.
• Observabilité : URL Spark History Server, répertoire des journaux d'événements, niveau de log, collecteur de métriques.

Commandes rapides :

spark-submit --version
pyspark --version
python -V
java -version
hdfs version

Dans un shell Spark ou un notebook :

sc.version
spark.version
spark.sparkContext.getConf().getAll()
spark.sparkContext.master

Prérequis :

• Accès à Spark UI et History Server (ou journaux d'événements) pour le job à optimiser.
• Autorité pour changer la configuration par job (spark-submit --conf, SparkSession builder, ou fichiers de config de job).
• Un échantillon de données représentatif qu'on peut relancer (exemple construit : échantillon de 5 % d'une table de 1 To stockée en Parquet).

Chemin de configuration sûr

Faites un changement à la fois, mesurez, puis décidez. Les étapes suivantes vont du faible risque et fort signal vers des changements plus profonds.

Étape 1 : Établir une ligne de base

• Choisissez un job unique représentatif (exemple construit : ETL nocturne joignant 200 Go Parquet + 5 Go tables de dimension, écriture sur HDFS).
• Enregistrez :
• Durée horloge et durées des étapes critiques.
• Temps de tâche maximum et médian ; indicateurs de déséquilibre (quelques tâches longues).
• Tailles de lecture/écriture de shuffle et événements de déversement (spill).
• Mémoire maximale des exécuteurs et temps de GC.
• Sauvegardez la config initiale : collectez les drapeaux spark-submit et configs SparkSession.

Astuce : activez les journaux d'événements s'ils ne le sont pas déjà.

--conf spark.eventLog.enabled=true \
--conf spark.eventLog.dir=hdfs:///spark-event-logs

Étape 2 : Bien dimensionner le parallélisme et les partitions

Symptômes : queues longues, beaucoup de tâches avec peu d'entrée, ou peu de partitions géantes.

• Définissez les partitions de shuffle par défaut pour charges SQL/DataFrame :

--conf spark.sql.shuffle.partitions=384

Exemple construit : 3x le total des cœurs exécuteurs (128 cœurs * 3 = 384).

• Pour charges RDD ou non-SQL, définissez :

--conf spark.default.parallelism=256

Exemple construit : 2x le total des cœurs.

• Ajustez le partitionnement d'entrée explicitement quand les sources ont de mauvais splits :

big = spark.read.parquet("data/fact
repartition(512)
dim = spark.read.parquet("data/dim
dim)
joined = big.join(dim, "key
dim)

• Coalescez seulement quand on réduit après opérations larges pour éviter trop de petits fichiers :

joined.coalesce(64).write.mode("overwrite
dim("out/path
dim

Résultats attendus : temps de tâche plus uniformes entre exécuteurs, queue longue réduite sur les étapes, et moins de fichiers de sortie si coalesce.

Étape 3 : Optimiser les jointures (broadcast, AQE et hints)

• Activez l'Exécution de Requête Adaptive (AQE - Adaptive Query Execution) pour coalescer les partitions de shuffle et choisir de meilleures stratégies de jointure à l'exécution :

--conf spark.sql.adaptive.enabled=true \
--conf spark.sql.adaptive.coalescePartitions.enabled=true

• Autorisez les jointures broadcast des petites tables, mais plafonnez la taille pour protéger la mémoire :

--conf spark.sql.autoBroadcastJoinThreshold=50MB

Seuil d'exemple construit.

• Quand vous savez qu'une table est petite et réutilisée souvent, utilisez un hint broadcast :

from pyspark.sql.functions import broadcast
joined = fact.join(broadcast(dim_small), "key
dim)

• Si du déséquilibre existe (quelques clés dominent), considérez le salage côté gros (exemple construit) :

from pyspark.sql.functions import rand, floor, col
salt_buckets = 8
fact_salted = fact.withColumn("salt", floor(rand() * salt_buckets))
dim_salted = dim_small.withColumn("salt", col("key
dim) % salt_buckets)
joined = fact_salted.join(dim_salted, ["key", "salt
dim).drop("salt
dim

Résultats attendus : données de shuffle réduites pour jointures petite dimension, moins de tâches déversées, étapes plus équilibrées sous déséquilibre.

Étape 4 : Sérialisation et compression

• Préférez Kryo pour la sérialisation d'objets JVM dans jobs RDD-lourds :

--conf spark.serializer=org.apache.spark.serializer.KryoSerializer

• Formats de données : utilisez formats colonnaires et compression splittable pour l'analytique :
• Parquet ou ORC avec Snappy pour vitesse et taille équilibrées.
• Considérez ZSTD pour compression plus forte avec CPU acceptable sur clusters modernes.
• Compression de shuffle :

--conf spark.shuffle.compress=true \
--conf spark.shuffle.spill.compress=true \
--conf spark.io.compression.codec=lz4

Résultats attendus : I/O plus faible et shuffles plus rapides sur jobs I/O-bound ; surveillez le CPU si passage à codecs plus lourds.

Étape 5 : Mettre en cache seulement ce qu'on réutilise

Mettez en cache les DataFrames utilisés plusieurs fois en aval ; sinon évitez le cache.

important = compute_expensive(df).persist()
# ... plusieurs actions sur `important`
important.unpersist()

Utilisez MEMORY_AND_DISK pour éviter l'OOM si le dataset ne tient pas entièrement en mémoire.

Étape 6 : Gérer le shuffle et l'I/O disque

• Assurez-vous que le service de shuffle externe est activé avec l'allocation dynamique (dépend du cluster).
• Augmentez prudemment les tailles réseau et tampons si forte attente de shuffle :

--conf spark.reducer.maxReqsInFlight=64 \
--conf spark.shuffle.file.buffer=64k \
--conf spark.shuffle.io.maxRetries=8 \
--conf spark.shuffle.io.retryWait=5s

• Si les déversements dominent, confirmez que les exécuteurs ont assez de mémoire et que les partitions ne sont pas surdimensionnées.

Étape 7 : Mémoire JVM, GC et dimensionnement des exécuteurs

• Visez 2-5 cœurs par exécuteur pour réduire les pauses GC et améliorer le parallélisme entre nœuds (ligne directrice construite).
• Équilibrez la mémoire :

--executor-cores 4 \
--executor-memory 8G \
--conf spark.executor.memoryOverhead=1536

• Préférez G1GC pour gros tas (JDK 11+). Définissez via options JVM si nécessaire :

--conf spark.executor.extraJavaOptions="-XX:+UseG1GC"

• Activez l'allocation dynamique pour charges mixtes avec un plafond pour éviter les voisins bruyants :

--conf spark.dynamicAllocation.enabled=true \
--conf spark.dynamicAllocation.minExecutors=8 \
--conf spark.dynamicAllocation.maxExecutors=128

Résultats attendus : moins de longues pauses GC, temps de tâche plus stables, parallélisme bien dimensionné sous charge.

Étape 8 : Optimisation de sortie et petits fichiers

• Consolidez les fichiers de sortie pour correspondre au parallélisme de lecture aval ; évitez des milliers de petits fichiers :

result.coalesce(128).write.mode("overwrite
dim("out
dim

• Pour tables partitionnées, assurez-vous que les colonnes de partition sont de cardinalité faible à moyenne et dimensionnées pour les requêtes typiques.

Tableau récapitulatif : leviers courants d'optimisation Spark

Vérification et diagnostics

Optimiser sans mesurer, c'est deviner. Validez chaque changement avec des signaux constants.

Utiliser Spark UI et History Server

• Jobs et étapes : visez les 2-3 étapes les plus longues par durée et shuffle.
• Tâches : regardez la distribution des durées de tâche. Quelques traînards indiquent déséquilibre ou goulots I/O distants.
• Onglet Storage : confirmez datasets mis en cache et usage mémoire.
• Onglet Environment : confirmez que vos nouvelles configs sont actives.

Contrôles latence et débit

• Latence : durée horloge du chemin critique (démarrage driver à fin dernière étape), plus durée d'étape au p95.
• Débit : enregistrements par seconde ou Mo/s aux étapes clés ; débit I/O lecture/écriture.
• Stabilité : temps GC en pourcentage du temps exécuteur et compteurs de retentatives.

Exemple construit (avant vs après AQE + partitionnement) :

• Avant : job 62 min, Étape 4 à 23 min avec 3 traînards à 6 min.
• Après : job 41 min, Étape 4 à 11 min, durées de tâche dans 10% de la médiane, zéro déversement.

SQL EXPLAIN et métriques

Lancez des plans détaillés pour repérer broadcasts et échanges :

df.explain(True)
# ou en SQL
spark.sql("EXPLAIN FORMATTED SELECT ...
dim).show(truncate=False)

Attendu : moins de nœuds Exchange après AQE ; BroadcastHashJoin pour petites tables ; nombre réduit de partitions de shuffle.

Journaux d'événements et GC

• Confirmez que les événements de déversement diminuent run après run en augmentant partitions ou mémoire.
• Cherchez les longues pauses GC ; visez temps GC < 5% du temps de tâche en régime stable (ligne directrice construite).
• Si vous changez la compression, surveillez temps CPU vs attente I/O dans métriques exécuteur (basculement CPU-bound vs I/O-bound).

Tableau de vérification : signaux et signification

Modes de défaillance et récupération

Mémoire insuffisante (exécuteur ou driver)

• Symptômes : ExecutorLostFailure, OutOfMemoryError, ou tâche tuée pendant broadcast.
• Correctifs rapides :
• Réduire seuil auto-broadcast ou retirer hints broadcast.
• Augmenter spark.executor.memoryOverhead pour gros shuffles.
• Passer cache à MEMORY_AND_DISK ou unpersist.
• Récupération : revenir à config snapshot antérieure ; relancer ligne de base pour confirmer stabilité.

Déséquilibre et queues longues

• Symptômes : quelques tâches tournent bien plus longtemps ; lectures shuffle fortement déséquilibrées.
• Correctifs :
• Augmenter partitions de shuffle modérément (ex. +25%).
• Saler clés côté gros ; pré-agréger côté déséquilibré.
• Broadcast côté petit pour éviter de le mélanger.
• Récupération : si performance dégrade, retirer salage et restaurer partitions antérieures.

Fichier shuffle introuvable / retentatives excessives

• Symptômes : échecs de récupération, retentatives de tâche répétées.
• Correctifs :
• Monter légèrement spark.shuffle.io.maxRetries et retryWait.
• Vérifier santé service shuffle externe ; vérifier pression disque local.
• Réduire taille partition pour limiter volume shuffle par tâche.
• Récupération : annuler réglages retentatives et changements partitions vers dernier bon connu.

Thrashing GC

• Symptômes : pics temps GC, faible débit tâche, GC complets fréquents.
• Correctifs :
• Moins de cœurs par exécuteur ; tas plus petits avec plus d'exécuteurs.
• G1GC et réglage cibles de pause si nécessaire.
• Sérialiseur Kryo pour code objet-lourd ; éviter grosses maps en mémoire.
• Récupération : restaurer dimensionnement exécuteur et réglages sérialiseur précédents.

Goulots côté driver

• Symptômes : long démarrage job, lent collect/show, pauses construction broadcast.
• Correctifs :
• Éviter collect driver sur gros datasets ; écrire vers stockage à la place.
• Augmenter mémoire driver si inévitable.
• Récupération : annuler mémoire driver et retirer actions côté driver.

Liste de contrôle opérationnelle

Quotidien ou par run de changement :

• Capturez la config courante depuis l'onglet Environment de Spark UI ; stockez un snapshot daté.
• Lancez le job représentatif sur un échantillon de données figé ; enregistrez durée, temps d'étape p95, lecture/écriture shuffle, compteurs déversement.
• Inspectez distribution temps de tâche ; confirmez absence de traînards.
• Confirmez statut AQE et nombre de partitions de shuffle effectives dans le plan.
• Vérifiez pourcentage temps GC exécuteur et pic mémoire.

Hygiène hebdomadaire :

• Revoyez comptes fichiers de sortie par table partitionnée et cibles de coalesce.
• Réévaluez seuils auto-broadcast vs tailles actuelles tables de dimension.
• Validez que bornes min/max allocation dynamique correspondent encore à la capacité du cluster.
• Purgez caches obsolètes dans applications Spark longue durée.

Gestion des changements :

• Un changement par run ; annotez notes de run avec le changement unique.
• Si pire, revenez immédiatement en arrière avec le snapshot de config sauvé.
• Gardez une petite bibliothèque de profils connus-bons (ex. profil ETL haut débit vs profil ad-hoc faible latence).

Profils de démarrage rapide (exemples construits)

Faible latence, volume de données modéré :

--conf spark.sql.adaptive.enabled=true \
--conf spark.sql.shuffle.partitions=200 \
--conf spark.sql.autoBroadcastJoinThreshold=80MB \
--executor-cores 3 --executor-memory 6G \
--conf spark.executor.memoryOverhead=1024 \
--conf spark.serializer=org.apache.spark.serializer.KryoSerializer

Haut débit, gros shuffles :

--conf spark.sql.adaptive.enabled=true \
--conf spark.sql.shuffle.partitions=800 \
--conf spark.shuffle.compress=true \
--conf spark.io.compression.codec=lz4 \
--executor-cores 4 --executor-memory 10G \
--conf spark.executor.memoryOverhead=2048 \
--conf spark.reducer.maxReqsInFlight=64

Conclusion

Vous avez maintenant un workflow sûr et incrémental pour optimiser Spark :

1. Établir une ligne de base avec un job unique représentatif.
2. Bien dimensionner les partitions.
3. Utiliser AQE et broadcasts sélectifs pour optimiser les jointures.
4. Ajuster sérialisation et compression pour votre profil I/O.
5. Dimensionner les exécuteurs pour contrôler GC et parallélisme.
6. Vérifier chaque changement avec Spark UI, EXPLAIN et journaux d'événements.
7. Garder les retours en arrière simples en changeant un levier à la fois.

Prochaines étapes :

• Packager un profil de base pour votre job le plus critique et vérifier sur un échantillon de données figé.
• Étendre aux jobs adjacents aux caractéristiques similaires, en ajustant seulement où les mesures le justifient.
• Automatiser la capture des métriques Spark UI et compteurs clés pour détecter les régressions avant les utilisateurs.

Avec une mesure disciplinée et des pas petits et réversibles, vous pouvez réduire la latence, augmenter le débit, et garder vos jobs Spark prévisibles alors que données et trafic grandissent.