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.

La raison d’être de la réoptimisation à l’exécution est qu’Azure Databricks dispose des dernières statistiques à la fin d’un échange de brassage et de diffusion (nommé étape de requête dans l’exécution adaptative des requêtes). Par conséquent, Azure Databricks peut opter pour une meilleure stratégie physique, choisir une taille et un nombre optimaux de partitions après le shuffle, ou effectuer des optimisations qui nécessitaient auparavant des indications, par exemple la gestion des jointures déséquilibrées.

Cela peut être très utile lorsque la collecte des statistiques n’est pas activée ou lorsque les statistiques sont obsolètes. Il est également utile dans les endroits où les statistiques dérivées statiquement sont inexactes, comme au milieu d’une requête complexe ou après l’occurrence d’un asymétrie des données.

Capacités

AQE est activé par défaut. Elle comporte quatre fonctionnalités 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étection et propagation dynamique des relations vides.

Application

AQE s’applique à toutes les requêtes qui sont :

• Requêtes qui ne sont pas des requêtes de streaming
• Contiennent au moins un échange (généralement lorsqu’il existe une jointure, un agrégat ou une fenêtre), une sous-requête ou les deux.

Toutes les requêtes appliquées à AQE ne sont pas nécessairement re-optimisées. La ré-optimisation peut ou ne pas apparaître avec un plan de requête différent de celui compilé statiquement. Pour déterminer si le plan d’une requête a été modifié par AQE, consultez la section suivante, Plans de requête.

Plans de requête

Cette section explique comment examiner les plans de requête de différentes façons.

Dans cette section :

• Interface utilisateur Spark
• DataFrame.explain()
• SQL EXPLAIN

Interface utilisateur Spark

Nœud AdaptiveSparkPlan

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. Avant l’exécution de la requête ou lorsqu’elle est en cours d’exécution, l’indicateur isFinalPlan du nœud correspondant AdaptiveSparkPlan s’affiche comme false; une fois l’exécution de la requête terminée, l’indicateur isFinalPlan passe à true.

Plan en évolution

Le diagramme du plan de requête évolue à mesure que l’exécution progresse et reflète le plan le plus actuel en cours d’exécution. Les nœuds qui ont déjà été exécutés (dans lesquels les métriques sont disponibles) ne changeront pas, mais ceux qui ne l'ont pas encore été peuvent changer au fil du temps à la suite de ré-optimisations.

Voici un exemple de diagramme de plan de requête :

DataFrame.explain()

Nœud AdaptiveSparkPlan

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. Avant l’exécution de la requête ou lorsqu’elle est en cours d’exécution, l’indicateur isFinalPlan du nœud correspondant AdaptiveSparkPlan s’affiche comme false; une fois l’exécution de la requête terminée, l’indicateur isFinalPlan passe à true.

Plan actuel et initial

Sous chaque AdaptiveSparkPlan nœud, il y aura à la fois le plan initial (le plan avant d’appliquer des optimisations AQE) et le plan actuel ou final, selon que l’exécution est terminée. Le plan actuel évoluera à mesure que l’exécution progresse.

Statistiques d’exécution

Chaque étape de remaniement et de diffusion contient des statistiques de données.

Avant l’exécution de l’étape ou lorsque l’étape est en cours d’exécution, les statistiques sont des estimations au moment de la compilation et l’indicateur isRuntime est false, par exemple : Statistics(sizeInBytes=1024.0 KiB, rowCount=4, isRuntime=false);

Une fois l’exécution de l’étape terminée, les statistiques sont celles collectées au moment de l’exécution, et l’indicateur isRuntime devient true, par exemple : Statistics(sizeInBytes=658.1 KiB, rowCount=2.81E+4, isRuntime=true)

Voici un DataFrame.explain exemple :

• Avant l’exécution

  

• Pendant l’exécution

  

• Après l’exécution

  

SQL EXPLAIN

Nœud AdaptiveSparkPlan

Les requêtes appliquées par AQE 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.

Aucun plan actuel

Comme SQL EXPLAIN il n’exécute pas la requête, le plan actuel est toujours identique au plan initial et ne reflète pas ce qui finirait par être exécuté par AQE.

Voici un exemple d’explication SQL :

Efficacité

Le plan de requête change si une ou plusieurs optimisations AQE prennent effet. L’effet de ces optimisations AQE est démontré par la différence entre les plans actuels et finaux et le plan initial et les nœuds de plan spécifiques dans les plans actuels et finaux.

• Modification dynamique de la jointure de fusion de tri en jointure hachée de diffusion : différents nœuds de jointure physique entre les plans actuel et final et le plan initial :

  

• Fusion dynamique des partitions : nœud CustomShuffleReader avec propriété Coalesced

  

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

• Gestion dynamique de la jointure d’asymétrie (nœud SortMergeJoin avec champ isSkew défini sur la valeur true) :

  

  

• Détecter et propager dynamiquement les relations vides : une partie (ou l’intégralité) du plan est remplacée par le nœud LocalTableScan avec le champ de la relation vide.

  

  

Paramétrage

Dans cette section :

• Activer et désactiver l’exécution de requêtes adaptatives
• Activer le shuffle optimisé automatiquement
• Modifier dynamiquement une jointure de fusion par tri en jointure de hachage de diffusion
• Fusionner dynamiquement les partitions
• Gérer dynamiquement la jointure biaisée
• Détecter et propager dynamiquement des relations vides

Activer et désactiver l’exécution de requêtes adaptatives

Activer le shuffle optimisé automatiquement

Modification dynamique de la jointure de fusion de tri en jointure hachée de diffusion

Fusion dynamique des partitions

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

Une partition est considérée comme asymétrique lorsque les deux (partition size > skewedPartitionFactor * median partition size) et (partition size > skewedPartitionThresholdInBytes) sont true.

Détecter et propager dynamiquement des relations vides

Questions fréquemment posées (FAQ)

Dans cette section :

• Pourquoi AQE n’a-t-il pas diffusé une petite table de jointure ?
• Dois-je encore utiliser un indicateur de stratégie de jointure de diffusion lorsque AQE est activé ?
• Quelle est la différence entre l’indicateur de jointure d’asymétrie et l’optimisation de jointure d’asymétrie de l’exécution adaptative des requêtes ? Lequel dois-je utiliser ?
• Pourquoi AQE n’a-t-il pas ajusté automatiquement l’ordre de mes jointures ?
• Pourquoi AQE n’a-t-il pas détecté mon asymétrie des données ?

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

Si la taille de la relation censée être diffusée est inférieure à ce seuil, mais n’est toujours pas diffusée :

• Vérifiez le type de jointure. La diffusion n’est pas prise en charge pour certains types de jointure, par exemple, la relation gauche d’un LEFT OUTER JOIN ne peut pas être diffusée.
• Il est également possible que la relation contienne un grand nombre de partitions vides. Dans ce cas, la majorité des tâches peut se terminer rapidement avec la jointure de fusion de tri ou peut potentiellement être optimisée avec la gestion des jointures d’asymétrie. L’exécution adaptative des requêtes évite la modification de ces jointures de fusion de tri pour diffuser des jointures hachées si le pourcentage de partitions non vides est inférieur à spark.sql.adaptive.nonEmptyPartitionRatioForBroadcastJoin.

Dois-je continuer à utiliser un indicateur de stratégie de jointure de diffusion avec l’exécution adaptative des requêtes activée ?

Oui. Une jointure de diffusion planifiée de manière statique est généralement plus performante qu’une jointure planifiée de manière dynamique par l’exécution adaptative des requêtes. En effet, celle-ci peut ne pas basculer vers la jonction de diffusion tant qu’elle n’a pas effectué de brassage pour les deux côtés de la jointure (la taille réelle des relations a à ce stade été récupérée). Ainsi, l’utilisation d’un indicateur de diffusion peut toujours être un bon choix si vous connaissez bien votre requête. AQE respecte les indicateurs de requête de la même façon que l’optimisation statique, mais peut toujours appliquer des optimisations dynamiques qui ne sont pas affectées par les indicateurs.

Quelle est la différence entre l’indicateur de jointure d’asymétrie et l’optimisation de jointure d’asymétrie de l’exécution adaptative des requêtes ? Lequel dois-je utiliser ?

Il est recommandé de s’appuyer sur la gestion des jointures d’asymétrie AQE plutôt que d’utiliser l’indicateur de jointure d’asymétrie AQE, car la jointure d’asymétrie AQE est complètement automatique et en général fonctionne mieux que l’équivalent de l’indicateur.

Pourquoi AQE n’a-t-il pas ajusté automatiquement l’ordre de mes jointures ?

La réorganisation dynamique des jointures ne fait pas partie de l'AQE.

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

Il existe deux conditions de taille qui doivent être satisfaites pour que L’AQE détecte une partition en tant que partition asymétrique :

• La taille de la partition est supérieure à la spark.sql.adaptive.skewJoin.skewedPartitionThresholdInBytes taille (par défaut, 256 Mo)
• La taille de la partition est supérieure à la taille médiane de toutes les partitions fois le facteur spark.sql.adaptive.skewJoin.skewedPartitionFactor de partition asymétrique (par défaut 5)

En outre, la gestion de l’asymétrie par AQE s’applique uniquement aux jointures basées sur un mélange (jointures de fusion par tri et jointures de hachage avec mélange). Les jointures de diffusion ne sont jamais optimisées pour l’asymétrie. Pour les jointures prises en charge, le type de jointure détermine quel côté AQE peut optimiser :

Par exemple, dans un LEFT OUTER JOIN, seul le décalage du côté gauche peut être optimisé, et un FULL OUTER JOIN n’est optimisé pour le décalage ni d’un côté ni de l’autre.

Héritage

Le terme « Exécution adaptative » existe depuis Spark 1.6, mais le nouvel AQE dans Spark 3.0 est fondamentalement différent. En termes de fonctionnalité, Spark 1.6 ne fait que la partie « partitions de fusion dynamique ». En termes d’architecture technique, le nouvel AQE est une infrastructure de planification dynamique et de replanification des requêtes basées sur des statistiques d’exécution, qui prend en charge diverses optimisations telles que celles que nous avons décrites dans cet article et peut être étendue pour permettre davantage d’optimisations potentielles.