AQE, DPP, skew join : ce que Spark 3 règle vraiment

Une montée de Spark 2 vers Spark 3 est souvent vendue avec un chiffre de gain. Le chiffre est atteignable, mais il ne vient pas de la version : il vient de trois mécanismes précis, qui ont chacun leurs conditions de déclenchement. Les connaître, c'est la différence entre un gain reproductible et un gain constaté une fois.

Le problème que résout AQE

Un optimiseur de requêtes planifie à partir de statistiques estimées avant l'exécution. Sur des données réelles, ces estimations sont fausses — un filtre sur une colonne corrélée, une jointure en cascade, et la cardinalité prédite s'écarte d'un ordre de grandeur.

Adaptive Query Execution change le moment de la décision : Spark découpe le plan aux frontières de shuffle, exécute une étape, observe les statistiques réelles des données produites, puis replanifie la suite. Ce n'est pas un meilleur devin, c'est un planificateur qui a le droit de changer d'avis.

spark.sql.adaptive.enabled = true                    # activé par défaut à partir de Spark 3.2
spark.sql.adaptive.coalescePartitions.enabled = true
spark.sql.adaptive.skewJoin.enabled = true
spark.sql.adaptive.advisoryPartitionSizeInBytes = 64m

Les trois optimisations, une par une

La fusion de partitions de shuffle. Le paramètre historique spark.sql.shuffle.partitions vaut 200 par défaut, quelle que soit la taille des données. Sur un petit volume, on obtient 200 tâches qui traitent quelques kilo-octets chacune : le coût d'ordonnancement dépasse le calcul. AQE mesure la taille réelle après le shuffle et fusionne les partitions pour approcher la taille conseillée. C'est le gain le plus régulier, et celui qu'on remarque le moins, parce qu'il s'exprime en tâches qui n'existent plus.

Le basculement de stratégie de jointure. Une jointure planifiée en sort-merge — donc avec deux shuffles — peut devenir un broadcast hash join si le côté réduit s'avère assez petit à l'exécution. C'est fréquent quand un filtre sélectif précède la jointure : le planificateur statique voyait une table de plusieurs gigaoctets, l'exécution en produit quelques dizaines de mégaoctets.

Le découpage des partitions déséquilibrées. C'est le mécanisme le plus visible sur les symptômes. Une jointure dont la clé a une valeur dominante concentre les lignes dans une seule partition : 199 tâches finissent en secondes, une tourne pendant quarante minutes, et le job entier attend. AQE détecte la partition anormalement grosse et la découpe en sous-partitions, en répliquant le côté opposé.

spark.sql.adaptive.skewJoin.skewedPartitionFactor = 5
spark.sql.adaptive.skewJoin.skewedPartitionThresholdInBytes = 256m

La condition est double : une partition est traitée comme déséquilibrée si elle dépasse threshold et si elle fait plus de factor fois la médiane. Un déséquilibre modéré mais généralisé ne déclenche donc rien — et c'est souvent celui-là qui coûte.

Dynamic Partition Pruning et ses conditions

DPP est indépendant d'AQE et répond à un autre problème : lire moins. Sur un schéma en étoile, on filtre la dimension et on joint la table de faits. Sans DPP, Spark lit toutes les partitions de la table de faits, puis jette ce qui ne joint pas.

DPP construit un filtre à partir des valeurs réellement retenues côté dimension, et le pousse dans la lecture de la table de faits. Le gain n'est pas marginal : on ne lit pas les fichiers.

-- DPP se déclenche
SELECT f.*
FROM   faits f
JOIN   dim d ON f.date_valeur = d.date_valeur   -- clé = colonne de partitionnement de f
WHERE  d.trimestre = '2026T1'                    -- filtre sélectif sur la dimension

Trois conditions doivent être réunies, et l'absence d'une seule suffit à annuler le bénéfice :

• la table de faits est partitionnée sur la colonne de jointure ;
• le filtre porte sur la dimension, pas sur la table de faits ;
• le côté dimension est broadcastable, ou spark.sql.optimizer.dynamicPartitionPruning.useStats permet l'estimation.

La première est celle qui manque le plus souvent, et pour une raison structurelle : les tables sont fréquemment partitionnées par date de chargement, alors que les requêtes filtrent par date de valeur. C'est un problème de modélisation, pas de moteur — et repartitionner la table de faits sur la colonne réellement interrogée produit un gain que ni AQE ni DPP ne peuvent apporter.

Ce qu'AQE ne règlera jamais

Il faut être précis sur la frontière, parce que c'est là que les migrations déçoivent.

Une clé de jointure massivement dégénérée. Si un tiers des lignes portent la même valeur — typiquement un code « inconnu » ou une valeur par défaut —, le découpage adaptatif atténue mais ne supprime pas le déséquilibre. La correction est en amont : isoler ces lignes et les traiter à part.

val nonRenseigne = lit("INCONNU")

val principal = faits.filter(col("id_tiers") =!= nonRenseigne)
                     .join(dim, Seq("id_tiers"))

val residuel  = faits.filter(col("id_tiers") === nonRenseigne)
                     .withColumn("libelle_tiers", lit(null: String))

val resultat = principal.unionByName(residuel)

Une explosion de cardinalité. Une jointure qui produit dix fois ses entrées reste coûteuse, quelle que soit la façon dont on répartit les partitions. Le sujet est la modélisation.

Un UDF opaque. Une fonction utilisateur est une boîte noire pour l'optimiseur : ni pushdown, ni élimination de colonnes. Remplacer un UDF par des fonctions natives est souvent plus rentable que tout le réglage adaptatif réuni.

Trop de petits fichiers. AQE agit sur les partitions de shuffle, pas sur la lecture initiale. Dix mille fichiers de 200 Ko produiront dix mille tâches de lecture avant qu'AQE n'ait son mot à dire.

Lire un plan avant de toucher un paramètre

La méthode qui évite de régler au hasard : lire le plan exécuté, pas le plan prévu. Avec AQE, les deux diffèrent, et c'est l'intérêt.

val df = requete
df.explain("formatted")     // plan initial
df.count()                  // exécution
// dans l'interface Spark, onglet SQL : le plan final porte les noeuds
// AdaptiveSparkPlan isFinalPlan=true, et les statistiques réellement observées

Ce qu'on cherche dans ce plan : un BroadcastHashJoin là où on attendait un SortMergeJoin, la présence d'un noeud de découpage de partitions déséquilibrées, et surtout le nombre d'octets réellement lus au niveau des scans de fichiers. Ce dernier chiffre dit si DPP a fonctionné — c'est la seule preuve qui vaille.

La démarche que j'applique, dans cet ordre : mesurer ce qui est lu, corriger le partitionnement si on lit trop, isoler les valeurs dégénérées si une tâche traîne, et seulement ensuite ajuster les paramètres adaptatifs. Dans cet ordre-là, les gains s'additionnent. Dans l'ordre inverse, on passe une semaine à régler des seuils sur un problème de modélisation.