wshobson/spark-optimization

Que fait spark-optimization ?

Optimisez les jobs Apache Spark avec le partitionnement, la mise en cache, l'optimisation du shuffle et le réglage de la mémoire. À utiliser pour améliorer les performances Spark, déboguer des jobs lents ou faire évoluer les pipelines de traitement de données.

Aperçu

Cette compétence vous aide à identifier et à résoudre les problèmes de performance courants dans les jobs Apache Spark. Elle fournit des conseils sur les techniques d'optimisation essentielles afin de garantir un traitement efficace des données.

Compétences clés

• Partitionnement : Ajustez le nombre de partitions en fonction de la taille des données et des ressources du cluster. Utilisez repartition() ou coalesce() pour contrôler le parallélisme. Évitez les small files en définissant spark.sql.shuffle.partitions de manière appropriée.
• Mise en cache : Mettez en cache les DataFrames ou les tables intermédiaires réutilisés à l'aide de df.cache() ou spark.catalog.cacheTable(). Utilisez df.unpersist() pour libérer la mémoire. Surveillez l'utilisation du stockage avec spark.sql.catalogImplementation.
• Optimisation du shuffle : Réduisez la surcharge du shuffle en utilisant le bucketing (bucketBy), sortWithinPartitions, ou coalesce après les opérations de groupBy/join. Ajustez la taille des fichiers de shuffle avec spark.sql.shuffle.partitions.
• Réglage de la mémoire : Allouez la mémoire de manière appropriée aux exécuteurs avec spark.executor.memory et spark.executor.memoryOverhead. Ajustez la mémoire de l'exécuteur JVM pour éviter les erreurs OutOfMemory.

Plates-formes prises en charge

• Databricks : Utilisez les clusters configurés pour l'optimisation automatique et la mise en cache Delta.
• Amazon EMR : Configurez Spark avec des paramètres de mémoire et de shuffle optimisés pour les instances EC2.
• Google Cloud Dataproc : Tirez parti des clusters éphémères et de la mise en cache de disque pour les charges de travail de big data.
• Clusters Spark autonomes : Appliquez ces optimisations dans n'importe quel environnement Spark local ou mutualisé.

Exemples d'utilisation

Exemple 1 : Optimiser un job de jointure

# Problème : Jointure lente en raison d'un nombre élevé de partitions
# Avant :
df1 = spark.read.parquet("s3://my-bucket/sales/")
df2 = spark.read.parquet("s3://my-bucket/products/")
result = df1.join(df2, "product_id")

# Après optimisation :
df1 = spark.read.parquet("s3://my-bucket/sales/")
df2 = spark.read.parquet("s3://my-bucket/products/")
df1_bucketed = df1.bucketBy(200, "product_id").saveAsTable("sales_bucketed")
df2_bucketed = df2.bucketBy(200, "product_id").saveAsTable("products_bucketed")
result = df1_bucketed.join(df2_bucketed, "product_id")  # Shuffle réduit

Exemple 2 : Mettre en cache des données intermédiaires

# Problème : Recalculer les mêmes données plusieurs fois
# Avant :
filtered_df = raw_df.filter(col("status") == "active")
agg1 = filtered_df.groupBy("category").count()
agg2 = filtered_df.groupBy("region").sum("sales")

# Après optimisation :
filtered_df = raw_df.filter(col("status") == "active").cache()
agg1 = filtered_df.groupBy("category").count()
agg2 = filtered_df.groupBy("region").sum("sales")
filtered_df.unpersist()  # Libérer la mémoire une fois terminé

Bonnes pratiques

1. Surveillez les métriques : Utilisez l'interface utilisateur Spark (port 4040) pour vérifier les temps de planification et d'exécution des tâches.
2. Adaptez-vous à la taille des données : Augmentez le nombre de partitions pour les grands jeux de données, diminuez-le pour les petits jeux.
3. Évitez les fuites mémoire : Désactivez toujours la mise en cache avec unpersist() quand ce n'est plus nécessaire.
4. Utilisez des formats de fichiers optimisés : Parquet, ORC, ou Avro avec compression Snappy réduisent les E/S.
5. Ajustez le nombre de partitions de shuffle : Commencez avec spark.sql.shuffle.partitions égal à 2-3 fois le nombre de cœurs du cluster.

Dépannage

• Problème : Erreur OutOfMemory : Augmentez spark.executor.memory et spark.executor.memoryOverhead. Réduisez le nombre de partitions avec coalesce().
• Problème : Small files : Utilisez repartition(n) ou coalesce(n) avant d'écrire, ou activez les fichiers compacts Delta.
• Problème : Shuffle excessif : Utilisez le bucketing ou sortWithinPartitions pour minimiser le shuffle lors des jointures ou des agrégations.

Références

• Documentation Apache Spark - Optimisation des performances
• Guide de réglage Databricks
• Bonnes pratiques EMR