Apache Spark est une puissante infrastructure informatique distribuée open source qui fournit une variété d’API pour prendre en charge le traitement du Big Data. PySpark est l’API Python pour Apache Spark, qui permet aux développeurs Python d’écrire des applications Spark en utilisant Python au lieu de Scala ou Java. De plus, les applications pySpark peuvent être ajustées pour optimiser les performances et obtenir un meilleur temps d’exécution, une meilleure évolutivité et une meilleure utilisation des ressources. Dans cet article, nous aborderons quelques astuces et techniques pour régler les applications PySpark.
1. Utiliser des variables de diffusion
Les variables de diffusion sont des variables en lecture seule qui peuvent être partagées entre les nœuds d’un cluster Spark. Les variables de diffusion peuvent être utilisées pour distribuer efficacement de grandes structures de données en lecture seule, telles que des tables de recherche, aux nœuds de travail. Cela peut réduire considérablement la surcharge du réseau et améliorer les performances. Dans PySpark, vous pouvez utiliser le broadcast fonction pour créer des variables de diffusion. Par exemple, pour diffuser une table de recherche nommée lookup_table:
from pyspark.sql.functions import broadcast
broadcast_table = broadcast(lookup_table)2. Utilisez des accumulateurs
Les accumulateurs sont des variables qui peuvent être utilisées pour accumuler des valeurs entre les nœuds d’un cluster Spark. Les accumulateurs peuvent être utilisés pour implémenter des fonctions d’agrégation personnalisées et collecter des statistiques sur les données en cours de traitement. Les accumulateurs ont des variables partagées qui sont mises à jour par les tâches exécutées sur les nœuds de travail et peuvent être lues par le programme du pilote. Dans PySpark, vous pouvez utiliser le SparkContext.accumulator méthode pour créer des accumulateurs. Par exemple, pour créer un accumulateur qui compte le nombre de lignes traitées :
from pyspark import SparkContext
sc = SparkContext() counter = sc.accumulator(0)
def process_row(row): # Process row counter.add(1)
data.map(process_row)
print("Number of rows processed:", counter.value)3. Utiliser la mise en cache RDD
La mise en cache RDD peut améliorer considérablement les performances en stockant les résultats intermédiaires en mémoire. Lorsqu’un RDD est mis en cache, Spark stocke les données en mémoire sur les nœuds de travail afin qu’elles soient accessibles plus rapidement. Cela peut réduire le temps consacré aux E/S disque et au recalcul des résultats intermédiaires. Dans PySpark, vous pouvez utiliser le RDD.cache() méthode pour mettre en cache un RDD. Par exemple:
4. Utiliser la mise en cache DataFrame
Les DataFrames sont une API de plus haut niveau que les RDD qui offrent une approche plus structurée du traitement des données. Les DataFrames peuvent être mis en cache pour améliorer les performances de la même manière que la mise en cache RDD. Dans PySpark, vous pouvez utiliser le DataFrame.cache() méthode pour mettre en cache un DataFrame. Par exemple:
5. Utilisez le format de fichier Parquet
Parquet est un format de fichier en colonnes optimisé pour le traitement de données volumineuses. Les fichiers Parquet peuvent être compressés pour réduire l’utilisation du disque et peuvent être lus et écrits plus efficacement que les autres formats de fichiers. Dans PySpark, vous pouvez utiliser le DataFrame.write.parquet() méthode pour écrire un DataFrame dans un fichier Parquet et la DataFrame.read.parquet() méthode pour lire un fichier Parquet dans un DataFrame. Par exemple:
df.write.parquet('path/to/parquet/file') parquet_df = spark.read.parquet('path/to/parquet/file')6. Utiliser le partitionnement
Le partitionnement est le processus de division des données en partitions, qui sont de plus petits sous-ensembles de données qui peuvent être traités indépendamment en parallèle. Spark utilise le partitionnement pour paralléliser le calcul et optimiser l’exécution du code. Lors de l’écriture du code PySpark, il est important de choisir un schéma de partitionnement approprié en fonction de la nature des données et des exigences de la tâche. Un bon schéma de partitionnement peut améliorer considérablement les performances en réduisant la surcharge du réseau et en minimisant le brassage des données. Dans PySpark, vous pouvez utiliser le DataFrame.repartition() méthode pour repartitionner un DataFrame.
7. Configurer les ressources du cluster
Le réglage des ressources de cluster est un élément essentiel de l’optimisation des performances de PySpark. Vous pouvez allouer des ressources telles que la mémoire et les cœurs de processeur à votre application en fonction de ses besoins. Pour allouer efficacement les ressources, vous pouvez utiliser les paramètres suivants :
spark.executor.instances: Ce paramètre définit le nombre d’exécuteurs à utiliser dans votre application.spark.executor.memory: Ce paramètre spécifie la quantité de mémoire à allouer à chaque exécuteur.spark.executor.cores: Ce paramètre définit le nombre de cœurs CPU à allouer à chaque exécuteur.
8. Optimiser la sérialisation
La sérialisation est le processus de conversion des données dans un format pouvant être transmis sur le réseau ou stocké sur un disque. PySpark utilise un format de sérialisation par défaut appelé Java Serialization, qui est lent et inefficace. Vous pouvez utiliser des formats de sérialisation plus efficaces comme Kryo ou Avro pour optimiser le processus de sérialisation et améliorer les performances de votre application.
Conclusion
Le réglage des applications PySpark nécessite une bonne compréhension des ressources du cluster et des exigences de l’application. En suivant les conseils mentionnés ci-dessus, vous pouvez optimiser les performances de vos applications PySpark et les rendre plus efficaces.