Techniques d’optimisation et Optimisation des Performances d’Apache Spark

Par Chandan Gaur / Big Data Engineering / 05, Oct, 2020

Qu’est-ce qu’Apache Spark ?

Apache, en 2012, a décrit la fondation Resilient Distributed Dataset (RDD dans Apache Spark) avec des jeux de données distribués en lecture seule sur des clusters distribués et l’a nommée Apache Spark. Plus tard, ils introduisent l’API Dataset, puis les API Dataframe pour la diffusion par lots et structurée de données. Cet article répertorie les meilleures techniques d’optimisation Apache Spark.

Apache Spark est une plate-forme de calcul de cluster rapide développée pour effectuer plus de calculs et de traitement de flux. Spark peut gérer une grande variété de charges de travail par rapport aux systèmes traditionnels qui nécessitent l’exécution et la prise en charge de plusieurs systèmes. Les pipelines d’analyse de données sont facilités par Spark en combinaison de différents types de traitement nécessaires à la production. Apache Spark est créé pour fonctionner avec un gestionnaire de cluster externe tel que YARN ou son gestionnaire autonome.

Caractéristiques d’Apache Spark

Certaines fonctionnalités d’Apache Spark incluent:-

• Plate-forme unifiée pour l’écriture d’applications Big Data.
• Facilité de développement.
• Conçu pour être très accessible.
• Spark peut fonctionner indépendamment. Cela donne donc de la flexibilité.
• Coût efficace.

XenonStack fournit des Services et des Solutions d’analyse pour l’Ingestion, le traitement et l’analyse rapides et efficaces des flux de données en temps réel et en flux pour l’Internet des Objets, la Surveillance, la Maintenance Préventive et Prédictive.

D’après l’article, Services de Streaming et d’Analyse en Temps Réel

Comprendre comment fonctionne l’optimisation Apache Spark ?

Afin de comprendre comment fonctionne l’optimisation Apache Spark, vous devez d’abord comprendre son architecture et dans la section suivante, nous allons élaborer la même chose.

L’architecture d’Apache Spark

L’architecture d’exécution de Spark se compose de trois parties –

Pilote Spark (processus maître)

Le pilote Spark convertit les programmes en tâches et planifie les tâches pour les exécuteurs. Le planificateur de tâches est la partie du pilote et aide à distribuer les tâches aux exécuteurs.

Gestionnaire de cluster Spark

Un gestionnaire de cluster est le cœur de Spark qui permet de lancer des exécuteurs et parfois des pilotes peuvent également être lancés par lui. Le planificateur Spark planifie les actions et les tâches dans l’application Spark de manière FIFO sur le gestionnaire de cluster lui-même. Vous devriez également lire sur Apache Airflow.

Exécuteurs (Processus esclaves)

Les exécuteurs sont les entités individuelles sur lesquelles une tâche individuelle d’un travail s’exécute. Les exécuteurs s’exécuteront toujours jusqu’au cycle de vie d’une application spark une fois qu’ils sont lancés. Les exécuteurs défaillants n’arrêtent pas l’exécution du travail spark.

RDD (Resilient Distributed Datasets)

Un RDD est une collection distribuée d’ensembles de données immuables sur des nœuds distribués du cluster. Un RDD est partitionné en une ou plusieurs partitions. RDD est le cœur de spark en tant que distribution entre les différents nœuds du cluster qui exploite la localité des données. Pour atteindre le parallélisme à l’intérieur de l’application, les partitions sont les unités pour celle-ci. Les transformations de répartition ou de fusion peuvent aider à maintenir le nombre de partitions. L’accès aux données est optimisé en utilisant le brassage RDD. Comme Spark est proche des données, il envoie des données à travers différents nœuds et crée les partitions requises selon les besoins.

DAG (Graphe acyclique dirigé)

Spark a tendance à générer un graphe opérateur lorsque nous entrons notre code dans la console Spark. Lorsqu’une action est déclenchée pour déclencher RDD, Spark soumet ce graphique au DAGScheduler. Il divise ensuite ces graphes d’opérateurs en étapes de la tâche à l’intérieur de DAGScheduler. Chaque étape peut contenir des tâches basées sur plusieurs partitions des données entrantes. Le DAGScheduler canalise ces graphes d’opérateurs individuels ensemble. Par exemple, les graphiques des opérateurs cartographiques planifient pour une seule étape et ces étapes sont transmises au . Planificateur de tâches dans le gestionnaire de cluster pour leur exécution. C’est la tâche du travail ou des exécuteurs d’exécuter ces tâches sur l’esclave.

Traitement distribué utilisant efficacement des partitions

L’augmentation du nombre d’exécuteurs sur les clusters augmente également le parallélisme dans le traitement du travail Spark. Mais pour cela, il faut disposer d’informations adéquates sur la façon dont ces données seraient distribuées entre ces exécuteurs via le partitionnement. RDD est utile dans ce cas avec un trafic négligeable pour le brassage des données entre ces exécuteurs. On peut personnaliser le partitionnement pour la paire RDD (RDD avec des paires clé-valeur). Spark assure que l’ensemble des clés apparaîtra toujours ensemble dans le même nœud car il n’y a pas de contrôle explicite dans ce cas.

La sécurité Apache Spark facilite l’authentification via un secret partagé. L’authentification Spark est le paramètre de configuration par lequel l’authentification peut être configurée. C’est le paramètre qui vérifie si les protocoles de la communication spark effectuent une authentification en utilisant un secret partagé ou non.

D’après l’article, Apache Spark Security

Erreurs à éviter lors de l’écriture des applications Spark

reduceByKey ou groupByKey

groupByKey et reduceByKey produisent la même réponse mais le concept pour produire des résultats est différent. reduceByKey est le mieux adapté aux grands ensembles de données car dans Spark, il combine la sortie avec une clé partagée pour chaque partition avant de mélanger les données. Tandis que de l’autre côté, groupByKey mélange toutes les paires clé-valeur. GroupByKey provoque des brassages inutiles et un transfert de données sur le réseau.

Maintenez la taille requise des blocs de lecture aléatoire

Par défaut, le bloc de lecture aléatoire Spark ne peut pas dépasser 2 Go. La meilleure utilisation est d’augmenter les partitions et de réduire sa capacité à ~ 128 Mo par partition, ce qui réduira la taille du bloc de lecture aléatoire. Nous pouvons utiliser la répartition ou la fusion dans des applications régulières. Les grandes partitions ralentissent le processus en raison d’une limite de 2 Go, et peu de partitions ne permettent pas de mettre à l’échelle le travail et d’atteindre le parallélisme.

Formats et délimiteurs de fichiers

Choisir les bons formats de fichiers pour chaque spécification liée aux données est un casse-tête. Il faut choisir judicieusement le format de données pour les types d’ingestion, le type intermédiaire et le type de sortie final. Nous pouvons également classer les formats de fichiers de données pour chaque type de plusieurs manières, comme nous pouvons utiliser le format de fichier AVRO pour stocker des données multimédias, car Avro est mieux optimisé pour les données binaires que Parquet. Parquet peut être utilisé pour stocker des informations de métadonnées car elles sont fortement compressées.

Petits fichiers de données

La diffusion est une technique permettant de charger de petits fichiers de données ou ensembles de données dans des blocs de mémoire afin qu’ils puissent être joints à des ensembles de données plus massifs avec moins de surcharge de brassage de données. Par exemple, Nous pouvons stocker de petits fichiers de données dans n nombre de blocs et de gros fichiers de données peuvent être joints à ces blocs de données à l’avenir, car de gros fichiers de données peuvent être répartis entre ces blocs de manière parallèle.

Aucune surveillance des étapes du travail

DAG est une structure de données utilisée dans Spark qui décrit différentes étapes des tâches sous forme de graphique. La plupart des développeurs écrivent et exécutent le code, mais la surveillance des tâches est essentielle. Ce suivi est mieux réalisé en gérant le DAG et en réduisant les étapes. Le travail avec 20 étapes est prolongé par rapport à un travail avec 3-4 étapes.

ByKey, repartition ou toute autre opération qui déclenche des mélangements

La plupart du temps, nous devons éviter autant que possible les mélangements de données, et parfois il devient très complexe d’obtenir une évolutivité à partir de ces mélangements. GroupByKey peut être un atout précieux, mais son besoin doit d’abord être décrit.

Apprentissage par renforcement

L’apprentissage par renforcement n’est pas seulement le concept permettant d’obtenir un meilleur environnement d’apprentissage automatique, mais également de mieux traiter les décisions. Il faut appliquer un apprentissage par renforcement profond dans spark si le modèle de transition et le modèle de récompense sont correctement construits sur des ensembles de données et si les agents sont suffisamment capables pour estimer les résultats.

Facteurs et techniques d’optimisation Apache Spark

L’une des meilleures caractéristiques de l’optimisation Apache Spark est son aide pour les calculs de données en mémoire. Le goulot d’étranglement pour ces calculs d’optimisation spark peut être le processeur, la mémoire ou toute ressource du cluster. Un besoin de sérialiser les données, de réduire la mémoire peut survenir dans de tels cas. Ces facteurs d’optimisation des étincelles, s’ils sont correctement utilisés, peuvent –

• Éliminer le processus de travail de longue durée
• Moteur d’exécution de correction
• Améliore le temps de performance en gérant les ressources

13 Techniques simples pour l’optimisation Apache Spark

Les accumulateurs sont des variables globales pour les exécuteurs qui ne peuvent être ajoutées que par une opération associative et commutative. Il peut donc être efficace en parallèle. Les accumulateurs peuvent être utilisés pour implémenter des compteurs (comme dans Map Reduce) ou une autre tâche telle que le suivi des appels d’API.

Par défaut, Spark prend en charge les accumulateurs numériques, mais les programmeurs ont l’avantage d’ajouter la prise en charge de nouveaux types. Spark garantit que la mise à jour de chaque tâche ne sera appliquée qu’une seule fois aux variables de l’accumulateur. Pendant les transformations, les utilisateurs doivent être conscients de la mise à jour de chaque tâche, car celles-ci peuvent être appliquées plus d’une fois si les étapes du travail sont réexécutées.

Performances de bucket Hive

Résultats de bucket avec un nombre fixe de fichiers car nous spécifions le nombre de buckets avec un bucket. Hive a pris le champ, calcule le hachage et attribue un enregistrement à ce compartiment particulier. Le bucket est plus stable lorsque le champ présente une cardinalité élevée, un traitement de données important et que les enregistrements sont répartis uniformément entre tous les compartiments, tandis que le partitionnement fonctionne lorsque la cardinalité du champ de partitionnement est faible.

Le bucket réduit la surcharge de tri des fichiers. Par exemple, si nous joignons deux tables qui contiennent un nombre égal de compartiments, spark rejoint directement les données en tant que compartiments de clés déjà triés. Le nombre de fichiers de compartiment peut être calculé comme plusieurs partitions en plusieurs compartiments.

Optimisation du pushdown des prédicats

Le pushdown des prédicats est une technique permettant de traiter uniquement les données requises. Les prédicats peuvent être appliqués à SparkSQL en définissant des filtres dans les conditions where. En utilisant la commande explain pour interroger, nous pouvons vérifier les étapes de traitement des requêtes. Si le plan de requête contient PushedFilter, la requête est optimisée pour sélectionner uniquement les données requises car chaque prédicat renvoie True ou False.

S’il n’y a pas de filtre poussé trouvé dans le plan de requête, mieux vaut lancer la condition where. Les poussées de prédicats limitent le nombre de fichiers et de partitions lus par SparkSQL lors de l’interrogation, réduisant ainsi les démarrages d’E/S de disque dans l’analyse en mémoire. Les requêtes sur des données dans des compartiments avec des poussées de prédicats produisent des résultats plus rapidement avec moins de lecture aléatoire.

Sérialisation/désérialisation sans données à l’aide d’Apache Arrow

Apache Arrow est utilisé comme format d’exécution en mémoire pour les moteurs de requêtes analytiques. Arrow fournit des brassages zéro de sérialisation / désérialisation des données via la mémoire partagée. Arrow flight envoie les grands ensembles de données sur le réseau. Arrow a son format de fichier arrow qui permet un accès aléatoire sans copie aux données sur les disques. Arrow dispose d’une couche d’accès aux données standard pour toutes les applications spark.

Il réduit la surcharge pour les opérations SerDe pour mélanger les données car il a un endroit commun où toutes les données résident et dans un format spécifique à la flèche.

Réglage du ramasse-miettes à l’aide de la collection G1GC

Lors du réglage des ramasseurs de miettes, nous recommandons d’abord d’utiliser G1 GC pour exécuter des applications Spark. Le garbage collector G1 gère entièrement les tas croissants qui sont couramment vus avec Spark. Avec G1, moins d’options seront nécessaires pour fournir à la fois un débit plus élevé et une latence plus faible. Pour contrôler les caractéristiques et les comportements imprévisibles de diverses applications, le réglage GC doit être maîtrisé en fonction des journaux générés.

Avant cela, d’autres techniques d’optimisation telles que le streaming et les solutions d’analyse en temps réel, dans la logique et le code du programme doivent être appliquées. La plupart du temps, G1GC permet d’optimiser le temps de pause entre les processus qui sont assez souvent dans les applications Spark, réduisant ainsi le temps d’exécution des tâches avec un système plus fiable.

Gestion et réglage de la mémoire

Comme nous le savons, pour des calculs tels que le brassage, le tri, etc., la mémoire d’exécution est utilisée alors qu’à des fins de mise en cache, on utilise une mémoire de stockage qui propage également des données internes. Il peut y avoir des cas où les travaux n’utilisent aucun cache ; par conséquent, des cas d’erreur d’espace pendant l’exécution. Les tâches mises en cache appliquent toujours moins d’espace de stockage où les données ne peuvent être expulsées par aucune exigence d’exécution. De plus, une application de streaming en temps réel avec Apache Spark peut être effectuée.

Nous pouvons définir spark.mémoire.fraction pour déterminer la quantité d’espace de tas JVM utilisée pour la mémoire d’exécution Spark. Généralement, 60% est la valeur par défaut. La mémoire de l’exécuteur doit être conservée le moins possible car elle peut entraîner un retard de la collecte des ordures de la JVM. Ce fait est également applicable aux petits exécuteurs car plusieurs tâches peuvent s’exécuter sur une seule instance JVM.

Localité des données

Dans Apache Spark, les tâches de traitement sont optimisées en plaçant le code d’exécution à proximité des données traitées, appelées localité des données. Parfois, la tâche de traitement doit attendre avant d’obtenir des données car les données ne sont pas disponibles. Cependant, quand le temps de l’étincelle.localité.wait expire, Spark essaie moins de niveau local, c’est-à-dire local au nœud pour le rack à n’importe quel.

Le transfert de données entre disques est très coûteux, de sorte que la plupart des opérations doivent être effectuées à l’endroit où résident les données. Cela permet de ne charger que de petites quantités de données, mais nécessite un développement basé sur les tests pour Apache Spark.

En utilisant des jointures colocalisées

Les jointures colocalisées prennent des décisions de redistribution et de diffusion. Nous pouvons définir de petits ensembles de données à localiser dans plusieurs blocs de mémoire pour une meilleure utilisation de la diffusion. Lors de l’application de jointures sur deux ensembles de données, spark trie d’abord les données des deux ensembles de données par clé et les fusionne.

Mais, nous pouvons également appliquer une clé de partition de tri avant de les joindre ou lors de la création de ces trames de données dans l’architecture de flèche d’INApache. Cela optimisera l’exécution de la requête car il n’y aurait pas d’appels de fonction inutiles à trier.

Mise en cache dans Spark

La mise en cache dans Apache Spark avec GPU est la meilleure technique pour l’optimisation Apache Spark lorsque nous avons besoin de données encore et encore. Mais il n’est toujours pas acceptable de mettre en cache des données.

Nous devons utiliser cache()RDD et DataFrames dans l’un des cas suivants –

• Quand il y a une boucle itérative comme dans les algorithmes d’apprentissage automatique.
• Lorsque RDD est consulté plusieurs fois dans une seule tâche ou tâche.
• Lorsque le coût de génération des partitions RDD est à nouveau plus élevé.

Cache() et persiste (StorageLevel.MEMORY_ONLY) peuvent être utilisés à la place les uns des autres. Chaque partition RDD qui est expulsée de la mémoire doit être à nouveau construite à partir de la source qui coûte encore très cher. L’une des meilleures solutions est d’utiliser persist (niveau de stockage.MEMORY_AND_DISK_ONLY) qui déverserait les partitions de RDD sur le disque local du Travailleur. Ce cas nécessite uniquement d’obtenir des données à partir du lecteur local du travailleur, ce qui est relativement rapide.

Taille de l’exécuteur

Lorsque nous exécutons des exécuteurs avec une mémoire élevée, cela entraîne souvent des retards excessifs dans la collecte des ordures. Nous devons garder le nombre de cœurs par exécuteur inférieur à cinq tâches par exécuteur. Des exécuteurs trop petits ne sont pas utiles en termes d’exécution de plusieurs tâches sur une seule machine virtuelle java.

Par exemple, les variables de diffusion doivent être répliquées pour chaque exécuteur exactement une fois, ce qui donnera plus de copies des données.

Fonction de fenêtrage Spark

Une fonction de fenêtre définit un cadre à travers lequel nous pouvons calculer les lignes d’entrée d’une table. Au niveau de la rangée individuelle. Chaque ligne peut avoir un cadre clair. Le fenêtrage nous permet de définir une fenêtre pour les données dans la trame de données. Nous pouvons comparer plusieurs lignes dans le même bloc de données.

Nous pouvons définir l’heure de la fenêtre sur un intervalle particulier qui résoudra le problème de dépendance des données avec les données précédentes. Le brassage dans Apache Beam est moindre sur les données précédemment traitées car nous conservons ces données pour l’intervalle de fenêtre.

Technique des filigranes

Le filigrane est une technique utile dans l’optimisation Apache Spark qui contraint le système par conception et aide à l’empêcher d’exploser pendant l’exécution. Le filigrane prend deux arguments –

• Colonne pour la durée de l’événement et
• Une durée de seuil qui spécifie la durée pendant laquelle nous devons traiter les données tardives

La requête dans l’architecture Apache Arrow sera automatiquement mise à jour si les données tombent dans ce seuil stipulé ; sinon, aucun traitement n’est déclenché pour ces données retardées. Il faut se rappeler que nous pouvons utiliser le mode complet côte à côte avec le filigrane car le mode complet conserve d’abord toutes les données dans la table résultante.

Sérialisation des données

L’optimisation Apache Spark fonctionne sur les données que nous devons traiter pour certains cas d’utilisation tels que l’analyse ou simplement pour le mouvement des données. Ce mouvement de données ou d’analyse peut être bien effectué si les données sont dans un format mieux sérialisé. Apache Spark prend en charge la sérialisation des données pour gérer efficacement les formats de données nécessaires à la Source ou à la Destination. Par défaut, Apache Spark utilise la sérialisation Java mais prend également en charge la sérialisation Kryo.

Par défaut, Spark utilise ObjectOutputStream de Java pour sérialiser les données. L’implémentation peut se faire via java.oi.Classe sérialisable. Il code les objets dans un flux d’octets. Il offre une persistance légère et flexible. Mais cela devient lent car cela conduit à d’énormes formats sérialisés pour chaque classe dans laquelle il est utilisé. Spark prend en charge la bibliothèque de sérialisation Kryo (v4) pour la sérialisation d’objets presque 10 fois plus rapide que la sérialisation Java car elle est plus compacte que Java.

Optimisation Apache Spark: Une approche globale

Apache Spark, un moteur de calcul distribué open source, est actuellement le framework le plus populaire pour le traitement par lots en mémoire, qui prend également en charge le streaming en temps réel. Avec son optimiseur de requêtes et son moteur d’exécution avancés, les techniques d’optimisation d’Apache Spark peuvent traiter et analyser de grands ensembles de données de manière très efficace. Cependant, l’exécution de techniques d’optimisation des jointures Apache Spark sans réglage minutieux peut dégrader les performances. Si vous souhaitez exploiter la puissance de votre application Apache Spark, consultez nos services Apache Spark gérés.

Catégories :

Développement de Big Data, Solutions Big Data, Analyse de données en Streaming,

Partager :