## Intro Des jobs Spark mal configurés gaspillent du calcul, bloquent des pipelines et masquent du potentiel de performance. Ce guide se concentre sur les erreurs les plus fréquentes, la manière de les diagnostiquer rapidement dans l’UI (User Interface) : interface utilisateur de Spark, et comment changer les paramètres en toute sécurité. Vous verrez des exemples concrets de spark-submit, un plan pilote reproductible à exécuter avant la production, et des rappels utiles pour les intégrations courantes avec Kafka, HDFS (Hadoop Distributed File System) : système de fichiers distribué Hadoop/stockage cloud, Airflow et NiFi. Ce que vous allez retenir : - Un flux de changement sécurisé pour modifier les configs Spark sans surprise - Les erreurs de configuration les plus courantes et comment les corriger - Des exemples concrets pour le batch et Structured Streaming - Un court chemin de dépannage réutilisable sur différents jobs ## Un flux de changement sécurisé Utilisez cette boucle pour toute modification de configuration : - Baseline - Capturez la version du job, la configuration complète (sc.getConf.getAll), la taille d’entrée, le temps d’exécution et les symptômes d’échec. - Sauvegardez les journaux d’événements (spark.eventLog.enabled=true) ou les liens du History Server depuis une exécution typique. - Hypothèse - Choisissez un seul changement visant un goulot mesurable (ex. : skew, taille de shuffle, temps de GC (Garbage Collection) : récupération de mémoire). - Pilote local - Testez sur un petit échantillon représentatif avec une graine ou une fenêtre temporelle fixe pour rendre les résultats reproductibles. - Rejeu en staging - Réexécutez la dernière fenêtre de production réussie avec la nouvelle config. Comparez le temps total, le skew des stages et la justesse (comptes de lignes et agrégats clés). - Déploiement gardé - Activez pour un sous-ensemble de jobs ou de partitions. Définissez des critères d’abandon (runtime +30 %, taux d’erreurs au-dessus du baseline, OOM (Out Of Memory) : épuisement de mémoire). - Monitoring - Surveillez dans l’UI : délai du scheduler, échecs de tâches, spill de shuffle, GC, débit IO (Input/Output) : entrées/sorties et rotation d’exécuteurs. - Revenir vite en arrière - Conservez les changements dans un seul bloc de config à rétablir en une étape. Utilisez les logs du driver/exécuteur et le History Server pour affiner la prochaine hypothèse. ## Diagnostics rapides dans l’UI Spark Ouvrez l’UI pendant l’exécution d’un job (ou le History Server après coup) et vérifiez : - Onglets Jobs/Stages : long délai du scheduler vs temps de tâche, tâches déséquilibrées, lecture/écriture de shuffle, spill (mémoire/disque) et retries de stages. - Onglet SQL (pour DataFrame/SQL) : stratégies de jointure, nombre de partitions de shuffle, coalescing adaptatif et indices de skew. - Onglet Executors : temps de GC, pic de mémoire de stockage, nombre d’échecs de tâches et exécuteurs black-listés. - Environnement : configuration effective à l’exécution. Règle pratique : si le délai du scheduler ou le spill de shuffle domine, corrigez le partitionnement ; si le temps de GC est élevé, ajustez mémoire/cœurs/overhead ; si quelques tâches durent beaucoup plus longtemps, traitez le skew. ## Erreurs courantes et correctifs pratiques ### 1) Déséquilibre entre exécuteurs et cœurs - Symptômes : faible utilisation CPU ou longues pauses de GC. - Correctif : dimensionnez cœurs et mémoire ensemble. Visez 1 tâche par cœur et 2–4 GiB de heap par cœur pour des workloads mixtes. - Exemple : spark-submit \ --conf spark.executor.instances=12 \ --conf spark.executor.cores=4 \ --conf spark.executor.memory=8g \ --conf spark.driver.memory=4g - Validation : l’onglet Executors montre un temps de tâche élevé vs délai du scheduler ; le temps de GC reste une faible fraction du temps de tâche. ### 2) Overhead mémoire trop faible (surtout avec PySpark ou des shuffles larges) - Symptômes : OOM (Out Of Memory) : épuisement de mémoire des conteneurs pendant le shuffle, OOM du worker Python, ou pertes fréquentes d’exécuteurs. - Correctif : augmentez l’overhead non-heap et la mémoire des workers Python si utilisée. - Exemple : spark-submit \ --conf spark.executor.memory=6g \ --conf spark.executor.memoryOverhead=1536 \ --conf spark.executor.pyspark.memory=1g - Validation : plus d’OOM dans les logs ; le spill de shuffle reste dans les limites mémoire des conteneurs ; nombre d’exécuteurs stable. ### 3) Mauvais réglage des partitions de shuffle - Symptômes : trop de petites tâches (overhead) ou trop peu de tâches énormes (OOM, skew). - Correctif : démarrez avec une cible de 128–256 Mo par partition de shuffle. Avec l’AQE (Adaptive Query Execution) : exécution adaptative des requêtes, fixez une borne supérieure raisonnable et laissez Spark coalescer. - Exemple : spark-submit \ --conf spark.sql.adaptive.enabled=true \ --conf spark.sql.shuffle.partitions=800 \ --conf spark.sql.adaptive.coalescePartitions.enabled=true \ --conf spark.sql.adaptive.advisoryPartitionSizeInBytes=256m - Validation : l’onglet SQL montre des partitions coalescées ; les durées de tâches sont homogènes avec un skew minimal. ### 4) AQE désactivée ou non ajustée - Symptômes : skews importants, shuffles inutiles ou stratégies de jointure inefficaces. - Correctif : activez l’AQE et la gestion du skew ; ajustez la taille de partition conseillée. - Exemple : spark-submit \ --conf spark.sql.adaptive.enabled=true \ --conf spark.sql.adaptive.skewJoin.enabled=true \ --conf spark.sql.adaptive.advisoryPartitionSizeInBytes=256m - Validation : le plan explain montre les stratégies de jointure attendues (BroadcastHashJoin ou SortMergeJoin) ; l’onglet SQL indique des partitions skewed scindées par l’AQE. ### 5) Allocation dynamique mal configurée - Symptômes : des exécuteurs inactifs persistent ou le job est à la diète sous charge. - Correctif : bornez la plage et fixez une valeur initiale raisonnable. Sur Kubernetes, préférez le suivi de shuffle ; sur YARN, utilisez l’external shuffle service. - Exemple (Kubernetes ou sans external shuffle service) : spark-submit \ --conf spark.dynamicAllocation.enabled=true \ --conf spark.dynamicAllocation.shuffleTracking.enabled=true \ --conf spark.dynamicAllocation.initialExecutors=8 \ --conf spark.dynamicAllocation.minExecutors=4 \ --conf spark.dynamicAllocation.maxExecutors=64 - Validation : le nombre d’exécuteurs suit la taille d’entrée ; le backlog côté scheduler reste faible ; inactivité limitée après les shuffles. ### 6) Broadcast joins surdimensionnés - Symptômes : OOM des exécuteurs pendant la jointure ou pression mémoire du driver lors de la collecte des données à diffuser. - Correctif : plafonnez la taille du broadcast ou désactivez-le quand il n’apporte pas de bénéfice ; envisagez des hints de jointure ou un repartitionnement. - Exemple : spark-submit \ --conf spark.sql.autoBroadcastJoinThreshold=64m \ --conf spark.sql.broadcastTimeout=1200 - Validation : le plan physique montre les stratégies de jointure attendues ; mémoire stable côté driver/exécuteurs. ### 7) Surcharge de sérialisation dans les pipelines RDD - Symptômes : CPU élevée en sérialisation et shuffles volumineux. - Correctif : préférez Kryo pour les classes personnalisées et enregistrez-les ; envisagez les API DataFrame/SQL pour profiter des encoders Tungsten. RDD (Resilient Distributed Dataset) : jeu de données distribué résilient. - Exemple : spark-submit \ --conf spark.serializer=org.apache.spark.serializer.KryoSerializer \ --conf spark.kryo.registrationRequired=true \ --conf spark.kryo.registrator=com.acme.support.MyRegistrator - Validation : baisse des tailles d’écriture/lecture de shuffle et tâches plus courtes vs Java serializer. ### 8) Exécution spéculative avec écritures lourdes ou sinks non idempotents - Symptômes : effets de bord dupliqués ou conflits côté sink (ex. : REST (Representational State Transfer) : API HTTP standard, JDBC (Java Database Connectivity) : connectivité Java aux bases de données, systèmes externes). - Correctif : désactivez la spéculation ou limitez-la aux tâches de fin de file sur des jobs en lecture seule. - Exemple : spark-submit --conf spark.speculation=false - Validation : pas d’écritures en double ; runtime inchangé sur des clusters stables. ### 9) Explosion de petits fichiers sur HDFS ou stockage cloud - Symptômes : de nombreux petits fichiers dégradent les listings de répertoires et les scans en aval. - Correctif : partitionnez et compacte(z) intentionnellement ; préférez repartition à coalesce pour les grands jeux de données. - Exemples : # Augmenter la taille de partition de lecture pour le scan de fichiers spark.conf.set("spark.sql.files.maxPartitionBytes", "256m") # Repartitionner par une clé stable et le temps, puis écrire final_df = transformed_df.repartitionByRange(200, "event_date", "user_id") final_df.write.mode("overwrite").partitionBy("event_date").parquet(path) # Lancer un job de compaction ultérieur si nécessaire spark.read.parquet(path).repartition(200).write.mode("overwrite").parquet(path) - Validation : moins de fichiers mais plus gros (ex. : 128–512 Mo) ; requêtes et listings en aval plus rapides. ### 10) Pression Kafka en Structured Streaming - Symptômes : latence de bout en bout croissante ou consumer lag qui ne se résorbe jamais. - Correctif : limitez l’entrée par trigger et redimensionnez correctement les partitions de shuffle ; checkpoint fiable. - Exemple : stream = (spark.readStream .format("kafka") .option("kafka.bootstrap.servers", servers) .option("subscribe", topics) .option("maxOffsetsPerTrigger", 500000) .load()) spark.conf.set("spark.sql.shuffle.partitions", 200) query = (process(stream) .writeStream .option("checkpointLocation", ckpt_path) .trigger(processingTime="30 seconds") .start()) - Validation : la durée des micro-batchs se stabilise ; le consumer lag reste borné malgré les pics. ### 11) Timeouts réseau trop agressifs pour de gros shuffles - Symptômes : exécuteur perdu, fetch failures ou retries intermittents de stages sur de grands shuffles. - Correctif : allongez le network timeout et le heartbeat pour des réseaux chargés ou bruyants. - Exemple : spark-submit \ --conf spark.network.timeout=800s \ --conf spark.executor.heartbeatInterval=60s - Validation : moins d’échecs de fetch de shuffle ; ensemble d’exécuteurs stable pendant les longs stages. ### 12) Surabonnement de l’orchestrateur (Airflow/NiFi) - Symptômes : surcharge du cluster quand plusieurs tâches de DAG ou processeurs NiFi démarrent en même temps. - Correctif : appliquez des limites de concurrence dans l’orchestrateur ; dimensionnez les exécuteurs selon le quota. - Exemples : - Airflow : utilisez max_active_runs_per_dag, des pools et la concurrency au niveau des tâches. - NiFi : définissez des seuils de back pressure et le nombre de tâches concurrentes par processeur. - Validation : temps d’attente acceptables ; pas de préemptions généralisées ni de stragglers. ## Liste de validation - Confirmez la config effective à l’exécution via l’onglet Environment ou sc.getConf.getAll. - Figez des métriques avant/après : temps total, temps stage/tâche, spill, GC, débit IO. - Protégez la justesse avec des comptes d’enregistrements et des agrégats clés (ex. : totaux par jour). - Gardez un seul changement par test pour isoler les effets. ## Plan pilote local Objectif : prouver une amélioration sur un job dans une exécution contrôlée, inspectable localement ou sur un cluster de dev avant déploiement. Périmètre - Choisissez un seul job batch ou une fenêtre de streaming. - Limitez-vous à une seule source de données (ex. : un topic Kafka ou un répertoire de stockage). - Améliorez une métrique (ex. : runtime, mémoire max d’exécuteur, dispersion du skew). Garde-fous - Fixez la taille d’entrée et la fenêtre temporelle. - Imposez une borne courte (2–4 exécutions). - Définissez des seuils de succès/échec et un déclencheur de rollback. Exemple d’étapes du pilote - Extrayez un échantillon représentatif de 5–10 Go (ou N minutes de données Kafka) vers un bucket de dev. - Appliquez un seul changement, comme spark.sql.shuffle.partitions=400 ou l’activation de l’AQE. - Exécutez spark-submit en local ou sur un cluster de dev avec un code identique mais la nouvelle conf. - Collectez le temps total, les métriques de skew par stage, les tailles de spill et les contrôles de justesse de sortie. - Décision : si le runtime s’améliore d’au moins 20 % et que la justesse tient, promouvez en staging ; sinon, revenez en arrière et testez l’hypothèse suivante. Exemples d’échantillonnage déterministe # Job batch basé sur des fichiers sample = df.sample(withReplacement=False, fraction=0.1, seed=42) # Fenêtre bornée dans le temps pour Kafka (Structured Streaming) stream = (spark.readStream.format("kafka") .option("kafka.bootstrap.servers", servers) .option("subscribe", topic) .option("startingOffsets", "earliest") .load()) Conseils opérationnels - Stockez les configs des jobs à côté du code avec un versionnage clair. - Imprimez sc.getConf.toDebugString au démarrage du job pour la traçabilité. - Étiquetez les logs avec un label de version de config pour des diff rapides. ## Parcours de dépannage Suivez ce chemin pour isoler rapidement les problèmes de configuration : - Identifiez la signature de l’échec - OOM, long délai du scheduler, stages déséquilibrés, GC élevé, fetch failures fréquents, ou filesystem lent. - Reproduisez sur un petit échantillon - Gardez le dataset constant ; désactivez temporairement les fonctionnalités adaptatives non liées pendant le debug pour isoler les causes. - Vérifiez la configuration effective - Déversez sc.getConf.getAll et comparez aux réglages visés ; capturez l’onglet Environment. - Inspectez l’UI Spark et les logs - Onglets Jobs/SQL : stages les plus longs, indices de skew, tailles d’écriture/lecture de shuffle. - Executors : temps de GC, pic de mémoire, échecs de tâches, blacklisting. - Logs driver/exécuteur : OOM de conteneur, erreurs de sérialisation, timeouts réseau. - Appliquez le plus petit changement viable - Ajustez un paramètre, réexécutez et mesurez les mêmes métriques. - Décidez et documentez - Si les métriques s’améliorent sans régression de justesse, conservez le changement. Sinon, rollback et testez la prochaine hypothèse. ## Conclusion La manière la plus rapide d’améliorer la fiabilité de Spark consiste à éviter quelques erreurs de configuration courantes et à changer les réglages avec discipline : baseline, pilote à un seul changement, rejeu en staging, déploiement gardé, et rollback instantané. Bâtissez des habitudes durables : maintenez un catalogue de configs versionné par job, journalisez la config effective à chaque exécution, suivez un petit jeu de métriques (runtime, skew, GC, IO) et gardez un rollback en une commande pour chaque changement. En cas de doute, testez d’abord en local, mesurez et privilégiez des pas petits et réversibles.