Exécution de requête adaptative

L’exécution de requête adaptative (AQE) est une nouvelle optimisation des requêtes qui se produit pendant l’exécution de la requête.

Avant qu’une requête ne soit lancée, l’optimiseur choisit un plan de requête basé sur des estimations de la taille des données. Ces estimations peuvent être erronées si les statistiques des tables manquent ou sont obsolètes, si les données sont déséquilibrées, ou pour les étapes ultérieures d’une requête complexe. Un plan construit sur de mauvaises estimations peut utiliser une stratégie de jointure lente, créer trop de petites tâches après un shuffle, ou créer quelques tâches surdimensionnées qui retardent l’ensemble de l’étape. AQE corrige ces problèmes en réoptimisant le plan pendant l’exécution de la requête, en fonction de la taille réelle des données.

À la fin de chaque échange de diffusion ou de regroupement, Azure Databricks dispose de statistiques précises pour les données produites jusqu’à présent. AQE utilise ces statistiques pour choisir une meilleure stratégie physique, choisir un nombre et une taille optimaux de partitions post-mélange, ou appliquer des optimisations qui nécessitent autrement des indications d’optimisation de requête, comme la gestion des jointures déséquilibrées.

AQE est activé par défaut. Vous n’avez pas besoin de modifier vos requêtes pour en bénéficier. Vérifiez si AQE a modifié un plan de requête lorsque vous dépannez une requête lente. Ajustez les réglages AQE uniquement lorsque les paramètres par défaut ne correspondent pas à votre charge de travail. Voir les plans de requête et configuration.

Capacités

L’AQE possède quatre caractéristiques majeures :

• Modifie dynamiquement la jointure de fusion de tri en jointure de synthèse de diffusion.
• Fusion dynamique des partitions (combinaison de partitions de petite taille en partitions de dimensions raisonnables) après un échange de brassage. Les petites tâches ont un débit d’E/S pire et ont tendance à souffrir davantage de la surcharge de planification et de la surcharge de configuration des tâches. Combiner de petites tâches économise les ressources et améliore le débit du cluster.
• Gestion dynamique de l’asymétrie dans la jointure de fusion de tri et la jointure hachée de brassage en fractionnant (et en répliquant si nécessaire) les tâches asymétriques en tâches de taille à peu près égale.
• Détecte et propage dynamiquement des relations vides, qui sont des parties d’une requête ne retournant aucune ligne.

Requêtes prises en charge

AQE s’applique aux requêtes qui remplissent ces deux conditions :

• La requête n’est pas une requête en streaming.
• La requête contient au moins un échange de données (généralement issu d’une jointure, d’un agrégat ou d’une fenêtre), au moins une sous-requête, ou les deux.

AQE réoptimise chaque requête prise en charge, mais le résultat peut être le même plan que celui que l’optimiseur avait compilé avant l’exécution de la requête. Pour vérifier si AQE a modifié le plan d’une requête, voir Plans de requête.

Plans de requête

Les requêtes auxquelles s’applique l’exécution adaptative des requêtes contiennent un ou plusieurs nœuds AdaptiveSparkPlan, généralement en tant que nœud racine de chaque requête principale ou sous-requête. Chaque AdaptiveSparkPlan nœud possède un isFinalPlan indicateur. Le drapeau est à false avant et pendant l’exécution de la requête, et passe à true à la fin de la requête.

Pour consulter un plan de requête, utilisez l’interface Spark, DataFrame.explain(), ou SQL EXPLAIN. Chacun montre des informations différentes sur la façon dont AQE modifie le plan.

Interface utilisateur Spark

Le diagramme du plan de requête dans l’interface Spark montre le plan le plus récent et se met à jour au fur et à mesure de l’exécution de la requête. Les nœuds qui ont déjà fonctionné et qui ont des métriques ne changent pas. Les nœuds qui n’ont pas encore été exécutés pourraient changer à mesure qu’AQE réoptimise le plan.

Cet exemple montre un plan de requête :

DataFrame.explain()

DataFrame.explain() affiche deux plans sous chaque nœud AdaptiveSparkPlan : le plan initial, avant qu’AQE n’applique toute optimisation, et le plan actuel. Le plan actuel change au fur et à mesure que la requête s’exécute et devient le plan final lorsque la requête se termine.

La sortie inclut des statistiques de données pour chaque phase de shuffle et de broadcast. Les statistiques dépendent du fait que l’étape a été exécutée ou non :

• Avant et pendant la course à l’étape, isRuntime est false et les statistiques sont des estimations au moment de la compilation. Par exemple : Statistics(sizeInBytes=1024.0 KiB, rowCount=4, isRuntime=false).
• Après l’exécution de l’étape, isRuntime est true et les statistiques sont des mesures de temps d’exécution. Par exemple : Statistics(sizeInBytes=658.1 KiB, rowCount=2.81E+4, isRuntime=true).

Ces captures d’écran montrent DataFrame.explain() la sortie avant, pendant et après l’exécution :

• Avant l'exécution

  

• Pendant l’exécution

  

• Après l’exécution

  

SQL EXPLAIN

SQL EXPLAIN ne lance pas la requête, donc AQE ne peut pas la réoptimiser. Le résultat ne montre que le plan initial, qui peut différer de celui qu’AQE exécute.

Cet exemple montre SQL EXPLAIN la sortie :

Identification des optimisations appliquées dans un plan de requête

Lorsque l’AQE applique une optimisation, le plan de requête change. Pour identifier quelles optimisations ont eu effet, comparez le plan actuel ou final avec le plan initial. Chaque optimisation modifie le plan d’une manière spécifique :

• Modification de la stratégie de jointure. Le plan actuel ou final a un nœud de jonction physique différent du plan initial.

  

• Fusion des partitions Le plan contient un nœud CustomShuffleReader avec la propriété Coalesced.

  

  < c0>Chaîne de lecture aléatoire personnalisée

• Gestion des skew joins. Le plan contient un nœud SortMergeJoin dont le champ isSkew est défini à true.

  

  

• Propagation de la relation vide. Un LocalTableScan nœud avec un champ de relation vide remplace une partie ou la totalité du plan.

  

  

Paramétrage

Ces propriétés de configuration Apache Spark contrôlent l’AQE. La plupart des charges de travail n’ont pas besoin de changer les paramètres par défaut.

Activez l’AQE et ajustez les optimisations individuelles

Utilisez ces propriétés pour activer ou désactiver AQE et pour régler le mélange automatique optimisé, la commutation des jointures broadcast et la propagation des relations vides :

Propriété	Type	Default	Description
spark.databricks.optimizer.adaptive.enabled	Boolean	true	Active ou désactive l’exécution adaptative de requêtes.
spark.sql.shuffle.partitions	Integer	200	Nombre par défaut de partitions à utiliser lors de la synchronisation des données pour les jointures ou les agrégations. Réglez-le sur auto pour activer le mélange automatique optimisé, qui définit ce chiffre en fonction du plan de requête et de la taille des données d’entrée de la requête. Pour le streaming structuré, vous ne pouvez pas changer cette valeur entre les redémarrages de requêtes depuis le même point de contrôle.
spark.databricks.adaptive.autoBroadcastJoinThreshold	Byte String	30MB	Le seuil de taille pour basculer une jointure de tri-fusion vers une jointure de hachage diffusée à l’exécution.
spark.databricks.adaptive.emptyRelationPropagation.enabled	Boolean	true	Activez ou désactivez la propagation dynamique des relations vides.

Fusion dynamique des partitions

Propriété	Type	Default	Description
spark.sql.adaptive.coalescePartitions.enabled	Boolean	true	Ça active ou désactive la fusion de partitions.
spark.sql.adaptive.advisoryPartitionSizeInBytes	Byte String	64MB	Taille cible après fusion. Les partitions coalescées sont proches, mais pas plus grandes que cette taille.
spark.sql.adaptive.coalescePartitions.minPartitionSize	Byte String	1MB	Taille minimale des partitions après fusion. Les partitions coalescées ne sont pas plus petites que cette taille.
spark.sql.adaptive.coalescePartitions.minPartitionNum	Integer	Deux fois le nombre de cœurs de cluster	Nombre minimal de partitions après fusion. Ne définissez pas cette propriété, car la définir remplace spark.sql.adaptive.coalescePartitions.minPartitionSize.

Gérer dynamiquement la jointure déséquilibrée

Propriété	Type	Default	Description
spark.sql.adaptive.skewJoin.enabled	Boolean	true	Ça active ou désactive la gestion des jointures asymétriques.
spark.sql.adaptive.skewJoin.skewedPartitionFactor	Integer	5	Le multiplicateur pour la taille médiane des partitions qu’une partition doit dépasser pour être considérée comme déséquilibrée.
spark.sql.adaptive.skewJoin.skewedPartitionThresholdInBytes	Byte String	256MB	La taille minimale qu’une partition doit dépasser pour être considérée comme déséquilibrée.

L’AQE considère une partition comme déséquilibrée seulement lorsque ces deux conditions sont vraies : (partition size > skewedPartitionFactor * median partition size) et (partition size > skewedPartitionThresholdInBytes).

Questions fréquemment posées (FAQ)

Pourquoi AQE n’a-t-il pas diffusé une petite table de jointure ?

Si le tableau est plus petit que spark.databricks.adaptive.autoBroadcastJoinThreshold mais qu’AQE ne le transmet pas en diffusion, vérifiez ces causes :

• Le type de jointure ne prend pas en charge le broadcast. Certains types de jointure ne peuvent pas diffuser l’un des côtés. Par exemple, AQE ne peut pas diffuser le côté gauche d’un LEFT OUTER JOIN.
• La table comporte de nombreuses partitions de table vides. La plupart des tâches dans la jointure de tri-fusion se terminent rapidement, ou la gestion du déséquilibre des jointures optimise la jointure. AQE ne transforme pas une jointure de tri-fusion en jointure de hachage broadcast si le pourcentage de partitions non vides est inférieur à spark.sql.adaptive.nonEmptyPartitionRatioForBroadcastJoin.

Dois-je quand même utiliser un Broadcast Join Hint avec AQE activé ?

Oui, si vous connaissez bien votre requête. Une jointure broadcast que l’optimiseur planifie avant l’exécution de la requête offre généralement de meilleures performances qu’une jointure qu’AQE planifie à l’exécution. AQE ne bascule peut-être pas vers une jointure broadcast avant d’avoir réparti les deux côtés de la jointure, moment où il apprend la taille réelle des tables.

AQE respecte les indications de requête de la même manière que l’optimisation statique, tout en appliquant les optimisations dynamiques non concernées par ces indications.

Dois-je utiliser le skew join hint ou le skew join handling AQE ?

Utilisez la gestion AQE des jointures asymétriques. C’est automatique et il fonctionne généralement mieux que l’indication de jointure asymétrique.

Pourquoi AQE n’a-t-il pas réorganisé mes jointures ?

AQE ne réorganise pas les jointures à l’exécution. Pour améliorer l’ordre de jointure, collectez des statistiques sur les tables afin que l’optimiseur basé sur le coût puisse choisir un meilleur plan. Voir Optimiseur basé sur les coûts.

Pourquoi AQE n’a-t-il pas détecté mon asymétrie des données ?

L’AQE considère une partition comme déséquilibrée uniquement lorsque ces deux conditions sont vraies :

• La partition est plus grande que spark.sql.adaptive.skewJoin.skewedPartitionThresholdInBytes (par défaut 256MB).
• La partition est plus grande que spark.sql.adaptive.skewJoin.skewedPartitionFactor (par défaut 5) fois la taille médiane de la partition.

AQE ne gère également le déséquilibre des données (skew) que dans les jointures basées sur le shuffle, qui sont des jointures de fusion par tri et des jointures de hachage avec shuffle. Il n’optimise jamais le skew lors des jointures broadcast. Pour les jointures basées sur le shuffle, le type de jointure détermine quel côté AQE peut optimiser :

Type de jointure	Inclinaison optimisée sur le côté gauche	Inclinaison optimisée sur le côté droit
INNER	Yes	Yes
CROSS	Yes	Yes
LEFT OUTER	Yes	Non
RIGHT OUTER	Non	Yes
LEFT SEMI	Yes	Non
LEFT ANTI	Yes	Non
FULL OUTER	Non	Non

Héritage

Spark 1.6 a introduit une fonctionnalité appelée exécution adaptative, mais AQE dans Spark 3.0 est une fonctionnalité différente. L’exécution adaptative de Spark 1.6 ne fusionne dynamiquement que les partitions. AQE est un framework qui planifie et replanifie les requêtes en fonction des statistiques d’exécution, et il prend en charge toutes les optimisations présentées sur cette page.