Techniques d’optimisation Spark – Perficient Blogs

Spark a été qualifié de « moteur de traitement de données distribué à usage général » et de « moteur d’analyse unifié ultra-rapide ».

Avec ses API simples, il simplifie la vie des développeurs et des programmeurs. Il est capable de gérer jusqu’à des pétaoctets de données et gère simultanément des milliers de machines virtuelles ou physiques.

Le framework Apache Spark est l’un des frameworks de calcul en cluster les plus populaires pour la gestion du Big Data. Cependant, lors de l’exécution de tâches Spark complexes, il est tout aussi important de prendre soin des performances. L’optimisation d’Apache Spark améliore les calculs en mémoire et augmente l’efficacité des tâches pour de meilleures caractéristiques de performances, en fonction de la distribution des données et de la charge de travail.

Certaines des techniques d’étincelle de base que vous pouvez adapter pour de meilleures performances incluent

• Cacher et persister
• Mélanger les partitions
• Stratégie de diffusion
• Sélection du format de fichier

Cacher et persister

L’exécution hautement efficace et parallélisée a un coût : il n’est pas facile de comprendre ce qui se passe à chaque étape. Vous devez collecter les données, ou un sous-ensemble de celles-ci, sur une seule machine afin de les visualiser après certaines transformations.

Spark doit donc déterminer le graphe de calcul, l’optimiser et effectuer cette activité. Cela peut prendre beaucoup de temps si votre ensemble de données est volumineux.

Cela prend particulièrement du temps, si la mise en cache est désactivée et que Spark doit lire les données d’entrée à partir d’une source distante, comme un cluster de base de données ou un stockage d’objets cloud comme S3.

Spark a ses propres mécanismes de mise en cache, tels que persist () et cache (). L’ensemble de données sera stocké en mémoire à l’aide de cache () et persist (). Lorsque vous avez un petit ensemble de données qui doit être utilisé à plusieurs reprises dans votre programme, nous cache-le.

Cache() : – Toujours en mémoire

Persist ():- Mémoire et disque

Si vous avez un ensemble de données limité qui est utilisé plusieurs fois dans votre programme, les mécanismes de persistance et de cache stockeront l’ensemble de données en mémoire chaque fois que nécessaire.

L’application de df.Cache() stockera toujours les données en mémoire, tandis que l’application de df.Persist() nous permettra de stocker certaines données en mémoire et d’autres sur le disque.

Par exemple:-

J’avais pris un exemple de données et je l’avais chargé en tant que dataframe.

Ensuite, j’applique des actions simples telles que compter dessus. Cela a pris environ 7 secondes.

Après la mise en cache, l’action a été effectuée en 0,54 seconde.

Supposons que je l’exécute avec des Go ou des To de données, cela prendra plusieurs heures pour terminer. Ainsi, Connaître ces concepts de base dans Spark permettrait donc de gagner plusieurs heures de calcul supplémentaires.

Mélanger les partitions

Le but des partitions aléatoires est de mélanger les données avant de les joindre ou de les agréger. Lorsque nous effectuons des opérations telles que group by, un brassage se produit. Pendant le brassage, une énorme quantité de données a été transférée d’une partition à une autre. Cela peut également se produire sur des partitions qui se trouvent sur la même machine ou entre des partitions qui se trouvent sur des machines différentes.

Lorsque vous traitez avec RDD, vous n’avez pas à vous soucier des partitions Shuffle. Supposons que nous fassions groupBy sur un RDD initial réparti sur 8 partitions. Malgré groupBy, le nombre de partitions reste inchangé dans RDD. Mais le cas n’est pas le même avec les trames de données.

Par exemple:-

> df = spark.createDataFrame(

[(‘1’, ‘true’),(‘2’, ‘false’),
(‘1’, ‘true’),(‘2’, ‘false’),
(‘1’, ‘true’),(‘2’, ‘false’),
(‘1’, ‘true’),(‘2’, ‘false’),
(‘1’, ‘true’),(‘2’, ‘false’),
])

> df.rdd.getNumPartitions()

8
#Opération groupBy en cours d’exécution

>group_df = df.groupBy(“_1”).count()

>group_df.show()

+—+—–+
| _1|compter|
+—+—–+
| 1| 5|
| 2| 5|
+—+—–+

>> group_df.rdd.getNumPartitions()

200

Le nombre de partitions aléatoires dans l’exemple ci-dessus était de 8, mais après l’application d’un groupBy, il a été augmenté à 200. En effet, la partition aléatoire Spark par défaut du DataFrame est de 200.

Le nombre de partitions spark shuffle peut être modifié dynamiquement avec la méthode conf dans la session Spark sparkSession.conf.Positionner(« spark.sql.shuffle.partitions »,100)

sparkSession.conf.set("spark.sql.shuffle.partitions",100)

ou Défini dynamiquement lors de l’initialisation de l’envoi de l’étincelle spark.sql.shuffle.partitions:100.

Le choix du nombre de partitions aléatoires approprié pour votre configuration Spark est crucial. Par exemple, supposons que je souhaite effectuer un groupBy sur un très petit ensemble de données en utilisant le nombre de partitions aléatoires par défaut de 200. Dans cette situation, je pourrais utiliser trop de partitions et surcharger mes ressources Spark.

Dans un scénario différent, j’ai un ensemble de données très volumineux et j’exécute un groupBy avec le nombre de partitions aléatoires par défaut. Donc, je pourrais ne pas utiliser toutes mes ressources d’étincelle.

Stratégie de diffusion

Lorsque nous joignons deux grands ensembles de données, ce qui se passe dans le backend est qu’une grande quantité de données est mélangée entre les partitions du même cluster ainsi qu’entre les partitions de différents exécuteurs. Lorsque nous devons joindre un jeu de données plus petit avec un jeu de données plus grand, nous effectuons des jointures de diffusion. Spark diffuse le plus petit ensemble de données à tous les nœuds du cluster lorsque nous appliquons une jointure de diffusion, de sorte que les données à joindre sont présentes sur tous les nœuds du cluster et Spark peut effectuer une jointure sans aucun mélange.

En utilisant la jointure de diffusion, vous pouvez éviter d’envoyer d’énormes données sur le réseau et de les mélanger. La méthode d’explication peut être utilisée pour valider si la trame de données a été diffusée ou non. Par défaut, la taille maximale d’une table à prendre en compte pour la diffusion est de 10 Mo. Elle est définie à l’aide de la variable spark.sql.autoBroadcastjoinThreshold. L’exemple ci-dessous illustre l’utilisation des jointures de diffusion.

Sélection du format de fichier

• Spark prend en charge une variété de formats, tels que CSV, JSON, XML, PARQUET, ORC, AVRO, etc.
• Choisir des fichiers de parquet avec une compression rapide peut optimiser les travaux Spark pour des performances élevées et une analyse optimale.
• Les fichiers Parquet sont natifs de Spark, qui contiennent des métadonnées avec leurs pieds de page.

Avec Spark, vous pouvez travailler avec un large éventail de formats de fichiers, tels que CSV, JSON, XML, PARQUET, ORC et AVRO. Il est possible d’optimiser le travail Spark en utilisant un fichier parquet qui a une compression rapide. Nous savons qu’un fichier parquet est natif de spark et est au format binaire. Outre les métadonnées, le fichier contient également un pied de page. Lorsque vous créez un fichier parquet, vous trouverez le fichier .metadata avec le fichier de données dans le même répertoire. L’exemple ci-dessous montre comment écrire un fichier parquet pour de meilleures performances.

DF = spark.read.json(« fichier.json »)

DF.write.parquet(« fichier.parquet »)

L’optimisation Spark peut être obtenue de plusieurs manières, si ces facteurs sont utilisés correctement pour des cas d’utilisation appropriés. nous pouvons :-

• Débarrassez-vous du processus de travail chronophage.

• améliorer la distribution des données et la charge de travail.

• Gérez les ressources afin d’optimiser les performances.

Cet article donne un aperçu de certains des aspects de base impliqués dans la création de travaux Spark efficaces. Dans la plupart des cas, la résolution des problèmes d’étincelles en suivant ces techniques résoudra le problème de manière efficace.

Bon apprentissage!!