Nous avons vu dans le précédent article : "Bonnes pratiques pour travailler avec Apache Spark sur Databricks – Partie 1" comment choisir des stratégies de jointure, quand utiliser la mise en cache/persistance, comment maximiser le parallélisme à grande échelle avec Apache Spark sur Databricks. Dans cet article, nous explorons plus en profondeur comment optimiser le plan de requête avec Adaptive Query Execution (AQE), et enfin comment surveiller avec Spark UI.

Optimiseur de catalyseur

L’optimisation Spark SQL améliore la productivité des développeurs et les performances des requêtes qu’ils écrivent, en utilisant des techniques telles que le filtrage précoce des données, l’utilisation des index disponibles et même la garantie que différentes sources de données sont jointes dans l’ordre le plus efficace.

Spark SQL Execution Plan

Au cœur de Spark SQL se trouve l’optimiseur Catalyst, qui utilise les puissantes fonctionnalités de Scala telles que la correspondance de modèles et la métaprogrammation d’exécution pour permettre aux développeurs de spécifier de manière concise des optimisations relationnelles complexes.

Il prend en charge l’optimisation basée sur des règles et sur les coûts.
Dans l’optimisation basée sur des règles, l’optimiseur basé sur des règles utilise un ensemble de règles pour déterminer comment exécuter la requête. Alors que l’optimisation basée sur les coûts trouve le moyen le plus approprié d’effectuer l’instruction SQL. Dans l’optimisation basée sur les coûts, plusieurs plans sont générés à l’aide de règles, puis leur coût est calculé, par exemple la phase de planification physique.

Exécution adaptative des requêtes (AQE)

Au fil des ans, des efforts considérables et continus ont été déployés pour améliorer le plan de requête de Spark SQL afin de générer des plans d’exécution de requête de haute qualité. Après l’intégration de l’optimiseur Catalyst dans Spark 2.0, Adaptive Query Execution est ajouté dans le prochain Spark 3.0. Il cherche à réoptimiser dynamiquement les plans de requête en fonction des statistiques d’exécution observées.
Cela rend beaucoup plus facile l’exécution de Spark car nous n’avons pas besoin de configurer ces choses à l’avance en tant que développeur, maintenant c’est la responsabilité d’AQE.

Dans Spark 3.0, l’infrastructure AQE est livrée avec trois fonctionnalités :
1. Fusion dynamique des partitions mélangées
2. Changement dynamique de stratégies de jointure
3. Optimisation dynamique des jointures asymétriques

Fusion dynamique des partitions mélangées
Lors de l’exécution de requêtes avec des données très volumineuses, le shuffle a généralement un impact très important sur les performances des requêtes, entre autres choses. Une propriété clé de shuffle est le nombre de partitions. Le meilleur nombre de partitions dépend des données.
Stratégie AQE : Définissez un nombre relativement important de partitions mélangées au début, puis combinez les petites partitions adjacentes en partitions plus grandes au moment de l’exécution en examinant les statistiques du fichier aléatoire.

Changement dynamique de stratégies de jointure

Spark prend en charge plusieurs stratégies de jointure, parmi lesquelles la jointure par hachage de diffusion est généralement la plus performante si un côté de la jointure peut bien s’intégrer dans la mémoire.
AQE peut replanifier la stratégie de jointure au moment de l’exécution en fonction de la taille de relation de jointure la plus précise. Par exemple, après avoir analysé la taille réelle des jeux de données, la jointure de fusion de tri est remplacée par la jointure de hachage de diffusion pour éviter les transformations inutiles.

Pour la jointure de hachage de diffusion convertie au moment de l’exécution, nous pouvons optimiser davantage la lecture aléatoire régulière en une lecture aléatoire localisée afin de réduire le trafic réseau.

Optimisation dynamique des jointures asymétriques

L’inclinaison des données se produit lorsque les données sont inégalement réparties entre les partitions du cluster. Une distorsion sévère peut réduire considérablement les performances des requêtes, en particulier avec les jointures.
AQE détecte automatiquement une telle distorsion à partir des statistiques des fichiers aléatoires.
Il divise ensuite les partitions asymétriques en sous-partitions plus petites, qui seront jointes à la partition correspondante de l’autre côté respectivement, ce qui se traduit par une meilleure performance globale.

Gérer dynamiquement la jointure d’inclinaison
spark.sql.adaptive.skewJoin.enabled

Fusionner dynamiquement les partitions
spark.conf.set(“spark.sql.adaptive.coalescePartitions.enabled“, “true”)

Interface utilisateur Spark
Spark UI offre diverses statistiques sur l’utilisation de la mémoire, les travaux, ainsi que les chronologies d’événements, les journaux. Ces statistiques peuvent vous donner un aperçu de ce qui se passe dans vos applications Spark. C’est aussi un bon outil pour détecter les problèmes de performances.

Une tâche Spark-submit lancera l’interface utilisateur Spark et vous pourrez vous y connecter sur l’hôte local ou via le pilote Spark sur le port par défaut 4040.

Spark UI – Emplois

L’onglet Travaux offre une vue de la chronologie des événements, indiquant quand les exécuteurs ont été ajoutés ou supprimés du cluster.

À partir de cette page de résumé, vous pouvez également accéder à une page de détails pour chaque tâche.

Par exemple, si la durée est élevée, c’est un bon indicateur que vous devriez rechercher quelle tâche pourrait causer des retards.

Spark UI – Étapes

L’onglet Étapes fournit un résumé de l’état actuel de toutes les étapes de toutes les tâches de l’application. Vous pouvez également accéder à une page de détails pour chaque étape, fournissant un DAG et des métriques sur ses tâches.

Spark UI – Exécuteurs

L’onglet Exécuteurs fournit des informations sur les exécuteurs créés pour l’application, par exemple, l’utilisation des ressources (disque, mémoire, cœurs), le temps passé dans GC, la quantité de données écrites et lues pendant le shuffle, etc.

En plus des statistiques récapitulatives, vous pouvez voir la quantité de mémoire utilisée par chaque exécuteur individuel et dans quel but. Cela permet également d’examiner l’utilisation des ressources lorsque vous avez utilisé la méthode cache() ou persist().

Spark UI – Stockage

L’onglet Stockage est souvent vide. Si les RDD ne sont pas persistants, c’est normal.
Il fournit des informations sur les tables ou les jeux de données mis en cache par l’application à la suite de la méthode cache() ou persist().

Interface utilisateur Spark – SQL

Les effets des requêtes Spark SQL exécutées dans le cadre de votre application Spark sont traçables et visibles via l’onglet SQL. Vous pouvez voir quand les requêtes ont été exécutées et par quelles tâches, et leur durée.

Ces métriques sont utiles lorsque nous voulons inspecter les détails d’un opérateur physique et découvrir ce qui s’est passé : combien de lignes ont été analysées, combien d’octets aléatoires ont été écrits, etc.

Spark UI – Environnement

L’onglet Environnement indique dans quel environnement votre application Spark s’exécute.
Il révèle de nombreux indices utiles pour le dépannage. p. ex.
1. Définition des propriétés spark,
2. Définition des propriétés système,
3. Environnement d’exécution (tel que JVM ou version Java) utilisé

En savoir plus avec <learning spark> Chapitre 7 « Optimisation et réglage de Spark pour l’efficacité »

Conclusion

Dans cet article, nous avons montré une série de techniques d’optimisation pour Spark sur Databricks. Vous serez maintenant en mesure de choisir la stratégie de jointure appropriée lors d’agrégations complexes, d’ignorer les opérations de remaniement coûteuses et améliorer le parallélisme par le partitionnement. Vous avez également eu un aperçu de deux outils puissants intégrés dans Spark 3.0 (optimiseur de catalyseur, exécution adaptative des requêtes) pour améliorer les performances des requêtes. Enfin, pour obtenir une perspective visuelle sur les performances, nous avons fait un tour d’horizon de l’interface utilisateur Spark.

References

[1] Learning spark O’REILLY Jules S. Damji, Brooke Wenig, Tathagata Das, and Denny Lee
[2] Databricks introduction, https://databricks.com/spark/about
[3] Databricks getting started with Apache Spark, https://databricks.com/spark/getting-started-with-apache-spark
[4] strategies of spark join, https://towardsdatascience.com/strategies-of-spark-join-c0e7b4572bcf
[5] optimize spark SQL join, https://medium.com/datakaresolutions/optimize-spark-sql-joins-c81b4e3ed7da
[6] spark SQL 3 common joins, https://www.linkedin.com/pulse/spark-sql-3-common-joins-explained-ram-ghadiyaram/
[7] joins in spark part 3, https://medium.com/@achilleus/https-medium-com-joins-in-apache-spark-part-3-1d40c1e51e1c
[8] how data partitioning in spark helps achieve more parallelism, https://www.dezyre.com/article/how-data-partitioning-in-spark-helps-achieve-more-parallelism/297
[9] Partitioning in Apache Spark, https://medium.com/@goyalsaurabh66/partitioning-in-apache-spark-7efece460f7c
[10] Databricks - Catalyst optimizer, https://databricks.com/glossary/catalyst-optimizer
[11] Databricks – adaptive query execution speeding up spark SQL at runtime, https://databricks.com/blog/2020/05/29/adaptive-query-execution-speeding-up-spark-sql-at-runtime.html