Aller au contenu principal

Module 8 — Réglage : partitions, mémoire, brassage des données

Le pipeline tourne, les résultats sont bons, mais l'entraînement prend sept heures là où il devrait en prendre deux. Ce module est celui du réglage — pas le réglage d'hyperparamètres du modèle (module 6), mais celui de Spark lui-même : partitions, mémoire, brassage. Les gains y sont souvent de deux à cinq, sans changer une ligne du modèle.

Nombre de partitions​

Deux nombres à distinguer. Le nombre de partitions à la lecture dépend du fichier : Spark découpe chaque fichier Parquet ou CSV en blocs, environ 128 Mo par défaut (spark.sql.files.maxPartitionBytes). Un dossier de 40 Go se lit donc dans 300 à 400 partitions initiales.

Le nombre de partitions après brassage est piloté par spark.sql.shuffle.partitions, qui vaut 200 par défaut. Ce chiffre convient à un jeu de 20 Go sur un petit cluster ; il devient absurde dans les deux directions extrêmes :

  • Sur un jeu de 1 Go, 200 partitions font des morceaux de 5 Mo. Le coût de planification écrase le calcul utile. Descendre à 8 ou 16.
  • Sur un jeu de 500 Go, 200 partitions font des morceaux de 2,5 Go chacune, souvent trop grosses pour la mémoire d'un exécuteur. Monter à 2 000 ou 4 000.

La règle simple : viser des partitions de 128 Mo à 512 Mo en sortie de brassage. spark.conf.set("spark.sql.shuffle.partitions", "800") se règle par travail, sans redémarrer la session.

repartition contre coalesce​

Deux fonctions pour changer le nombre de partitions d'un DataFrame, avec des coûts très différents :

  • repartition(n) déclenche un brassage complet : chaque ligne est redistribuée uniformément sur n partitions. Coûteux, mais produit des partitions équilibrées. Sert pour augmenter le nombre de partitions, ou pour équilibrer une répartition biaisée.
  • coalesce(n) fusionne les partitions existantes sans brassage, simplement en combinant plusieurs partitions en une. Très bon marché, mais ne peut que diminuer le nombre de partitions, et peut produire des tailles très inégales.

Cas d'usage classiques :

# Écriture finale d'un rapport agrégé : 12 lignes réparties sur 200 partitions
resultat.coalesce(1).write.parquet("resultat.parquet")     # un seul fichier

# Répartition avant un algorithme dont on veut équilibrer la charge
donnees_equilibrees = donnees.repartition(64, "compagnie")

repartition(n, col) groupe par la colonne en n partitions selon un hachage — utile pour joindre ensuite sur cette colonne sans nouveau brassage.

L'asymétrie des clés​

Le cas qui coûte le plus cher n'est pas d'avoir trop peu ou trop de partitions : c'est d'avoir des partitions très inégales, autrement dit une asymétrie (data skew). Un groupBy("compagnie") sur un jeu où Air France pèse 40 % des vols produit une partition Air France énorme, cinq partitions moyennes, et cent partitions minuscules. Les exécuteurs traitant les petites partitions terminent en quelques secondes ; celui qui a Air France travaille pendant deux heures. Le travail entier attend cet exécuteur, gaspillant tout le reste de la grappe.

Trois approches, à empiler si nécessaire :

  1. Activer l'exécution adaptative des requêtes (AQE, activée par défaut depuis Spark 3.2), qui repère les partitions asymétriques et les scinde automatiquement pendant l'exécution : spark.conf.set("spark.sql.adaptive.skewJoin.enabled", "true").
  2. Utiliser un broadcast join quand la table de droite d'une jointure tient en mémoire d'un exécuteur (moins de quelques centaines de Mo) : F.broadcast(petite_table). Aucun brassage sur la grosse table.
  3. En dernier recours, saler manuellement la clé : ajouter un suffixe aléatoire pour disperser les lignes du géant sur plusieurs partitions, puis regrouper après le calcul.

Mémoire des exécuteurs​

Deux paramètres pilotent la mémoire d'un exécuteur JVM : spark.executor.memory (mémoire heap) et spark.executor.memoryOverhead (hors heap, pour le code natif et Python en PySpark). En PySpark, la mémoire consommée par Python vit dans l'overhead — la sous-estimer provoque des morts d'exécuteurs avec le message tristement célèbre ExecutorLostFailure : Container killed by YARN for exceeding memory limits.

Règles utiles :

  • Un exécuteur de 4 à 8 Go tourne mieux que trois exécuteurs de 32 Go : moins de pression sur le ramasse-miettes, meilleure parallélisation.
  • spark.executor.cores typiquement entre 2 et 5. Au-delà, la contention Java sur les allocations mémoire annule le gain.
  • En PySpark, prévoir 15 à 20 % de mémoire supplémentaire en overhead.

Cache, oui mais avec discernement​

Rappel du module 3 : cache() ne se justifie que si le DataFrame est relu au moins deux fois. Le cache lui-même consomme de la mémoire d'exécuteur ; y stocker 40 Go pour n'en relire qu'une fois 500 Mo est une nuisance. Toujours accompagner un cache() d'un unpersist() lorsque son usage se termine, sinon la mémoire reste occupée jusqu'à la fin de la session.

Lire l'interface Spark​

L'interface web sur http://localhost:4040 (ou celle du cluster Databricks / YARN) contient tout ce qu'il faut pour diagnostiquer. Les onglets utiles au quotidien :

  • Jobs : chaque action est un Job ; sa durée totale et sa décomposition en étapes.
  • Stages : c'est là qu'on voit l'asymétrie. Une étape avec un histogramme de durées de tâches où quelques barres dépassent d'un facteur cent est le symptôme évident.
  • SQL / DataFrame : le plan physique de chaque requête, avec les tailles lues et les brassages.
  • Storage : ce qui est actuellement en cache et combien de mémoire il consomme.
  • Executors : mémoire utilisée par exécuteur, pertes éventuelles, logs d'erreur détaillés.
Le petit fichier

Un dossier Parquet écrit après un traitement contient parfois un fichier par partition en sortie, soit 200 fichiers de quelques mégaoctets si spark.sql.shuffle.partitions=200. La lecture aval devient lente à cause du surcoût d'ouverture. df.coalesce(20).write.parquet(...) avant l'écriture ramène à 20 fichiers de taille raisonnable.

En résumé​

  • spark.sql.shuffle.partitions=200 est un défaut qu'il faut presque toujours revoir ; viser des partitions de 128 Mo à 512 Mo.
  • repartition brasse et équilibre ; coalesce fusionne sans brasser mais peut déséquilibrer.
  • L'asymétrie des clés est le pire ennemi : AQE, broadcast join, salage manuel.
  • Petits exécuteurs, cache mesuré, interface Spark comme premier réflexe de diagnostic.

Module suivant : sauvegarder le pipeline entraîné et écrire les résultats pour la consommation en aval.