F. Courouge
Avancé
⭐ Article vedette

Apache Spark : Optimisation et Tuning pour la Production

Guide complet pour optimiser Apache Spark en production : configuration, partitioning, caching, et techniques avancées pour maximiser les performances.

Publié le
16 décembre 2024
Lecture
20 min
Vues
0
Auteur
Florian Courouge
Spark
Big Data
Performance
Scala
Python
Optimization

Table des matières

📋 Vue d'ensemble rapide des sujets traités dans cet article

Cliquez sur les sections ci-dessous pour naviguer rapidement

Apache Spark : Optimisation et Tuning pour la Production

Apache Spark est devenu l'un des frameworks de traitement de données les plus populaires, mais ses performances dépendent fortement de sa configuration et de son utilisation. Ce guide vous accompagnera dans l'optimisation de Spark pour la production.

💡Architecture et Concepts Fondamentaux

Comprendre l'Architecture Spark

Driver Program

  • •Point d'entrée de l'application
  • •Gestion du SparkContext
  • •Planification des tâches
  • •Collecte des résultats

Cluster Manager

  • •YARN, Mesos, Kubernetes, Standalone
  • •Allocation des ressources
  • •Gestion des executors

Executors

  • •Processus JVM sur les worker nodes
  • •Exécution des tâches
  • •Stockage des données en cache
  • •Communication avec le driver

Tasks et Stages

  • •Unité de travail la plus petite
  • •Regroupement par transformations
  • •Optimisation par le Catalyst Optimizer

Métriques Clés à Surveiller

// Configuration de monitoring
val spark = SparkSession.builder()
  .appName("OptimizedSparkApp")
  .config("spark.eventLog.enabled", "true")
  .config("spark.eventLog.dir", "/var/log/spark-events")
  .config("spark.history.fs.logDirectory", "/var/log/spark-events")
  .config("spark.sql.adaptive.enabled", "true")
  .config("spark.sql.adaptive.coalescePartitions.enabled", "true")
  .getOrCreate()

// Métriques importantes à surveiller
import org.apache.spark.sql.execution.ui.SparkListenerSQLExecutionStart
import org.apache.spark.scheduler.SparkListener

class PerformanceListener extends SparkListener {
  override def onTaskEnd(taskEnd: SparkListenerTaskEnd): Unit = {
    val metrics = taskEnd.taskMetrics
    println(s"Task Duration: ${taskEnd.taskInfo.duration}ms")
    println(s"GC Time: ${metrics.jvmGCTime}ms")
    println(s"Shuffle Read: ${metrics.shuffleReadMetrics.totalBytesRead} bytes")
    println(s"Shuffle Write: ${metrics.shuffleWriteMetrics.bytesWritten} bytes")
  }
}

spark.sparkContext.addSparkListener(new PerformanceListener())

💡Configuration et Tuning des Ressources

Dimensionnement des Executors

# Configuration optimale pour un cluster avec 6 cores par node et 64GB RAM

# Approche 1: Fat Executors (non recommandé)
spark-submit \
  --executor-cores 5 \
  --executor-memory 63g \
  --num-executors 2

# Approche 2: Tiny Executors (non recommandé)
spark-submit \
  --executor-cores 1 \
  --executor-memory 10g \
  --num-executors 10

# Approche 3: Balanced (recommandé)
spark-submit \
  --executor-cores 3 \
  --executor-memory 19g \
  --executor-memory-fraction 0.8 \
  --num-executors 4 \
  --driver-memory 4g \
  --driver-cores 2

Configuration JVM et Garbage Collection

# Configuration JVM optimisée
spark-submit \
  --conf "spark.executor.extraJavaOptions=-XX:+UseG1GC -XX:+UnlockDiagnosticVMOptions -XX:+G1PrintRegionRememberSetInfo -XX:+PrintGC -XX:+PrintGCDetails -XX:+PrintGCTimeStamps -XX:+PrintGCApplicationStoppedTime -Xloggc:/var/log/spark/gc-executor.log" \
  --conf "spark.driver.extraJavaOptions=-XX:+UseG1GC -XX:+UnlockDiagnosticVMOptions -XX:+G1PrintRegionRememberSetInfo -XX:+PrintGC -XX:+PrintGCDetails -XX:+PrintGCTimeStamps -XX:+PrintGCApplicationStoppedTime -Xloggc:/var/log/spark/gc-driver.log" \
  --conf "spark.executor.memory=19g" \
  --conf "spark.executor.memoryFraction=0.8" \
  --conf "spark.storage.memoryFraction=0.3"

Configuration Réseau et Sérialisation

// Configuration optimisée pour la sérialisation
val spark = SparkSession.builder()
  .config("spark.serializer", "org.apache.spark.serializer.KryoSerializer")
  .config("spark.kryo.registrationRequired", "false")
  .config("spark.kryo.unsafe", "true")
  .config("spark.kryoserializer.buffer.max", "1024m")
  .config("spark.rdd.compress", "true")
  .config("spark.broadcast.compress", "true")
  .config("spark.shuffle.compress", "true")
  .config("spark.shuffle.spill.compress", "true")
  .getOrCreate()

// Enregistrement des classes pour Kryo
spark.conf.set("spark.kryo.registrator", "com.example.MyKryoRegistrator")

class MyKryoRegistrator extends KryoRegistrator {
  override def registerClasses(kryo: Kryo): Unit = {
    kryo.register(classOf[MyCustomClass])
    kryo.register(classOf[scala.collection.mutable.WrappedArray.ofRef[_]])
  }
}

💡Optimisation du Partitioning

Stratégies de Partitioning

import org.apache.spark.sql.functions._
import org.apache.spark.sql.types._

// Lecture avec partitioning optimal
val df = spark.read
  .option("multiline", "true")
  .option("maxFilesPerTrigger", "1000")
  .parquet("/data/input")

// Vérifier le nombre de partitions
println(s"Nombre de partitions: ${df.rdd.getNumPartitions}")

// Repartitioning intelligent
val optimalPartitions = (df.count() / 100000).toInt.max(200).min(2000)
val repartitionedDF = df.repartition(optimalPartitions)

// Partitioning par colonne pour les jointures
val partitionedByKey = df.repartition(col("user_id"))

// Coalesce pour réduire les petites partitions
val coalescedDF = df.coalesce(100)

// Fonction utilitaire pour analyser les partitions
def analyzePartitions(df: DataFrame): Unit = {
  val partitionSizes = df.rdd.mapPartitions { iter =>
    Iterator(iter.size)
  }.collect()
  
  println(s"Partition count: ${partitionSizes.length}")
  println(s"Min partition size: ${partitionSizes.min}")
  println(s"Max partition size: ${partitionSizes.max}")
  println(s"Avg partition size: ${partitionSizes.sum / partitionSizes.length}")
  println(s"Std deviation: ${math.sqrt(partitionSizes.map(x => math.pow(x - partitionSizes.sum / partitionSizes.length, 2)).sum / partitionSizes.length)}")
}

analyzePartitions(df)

Optimisation des Jointures

// Broadcast Join pour les petites tables
val smallTable = spark.read.parquet("/data/small_table")
val largeTable = spark.read.parquet("/data/large_table")

// Force broadcast join
val result = largeTable.join(
  broadcast(smallTable),
  largeTable("key") === smallTable("key")
)

// Bucketing pour optimiser les jointures répétées
largeTable.write
  .bucketBy(200, "user_id")
  .sortBy("timestamp")
  .option("path", "/data/bucketed_table")
  .saveAsTable("bucketed_table")

// Salt technique pour éviter le data skew
def saltedJoin(left: DataFrame, right: DataFrame, joinKey: String, saltFactor: Int = 100): DataFrame = {
  val saltedLeft = left.withColumn("salt", (rand() * saltFactor).cast("int"))
    .withColumn("salted_key", concat(col(joinKey), lit("_"), col("salt")))
  
  val saltedRight = right.withColumn("salt", explode(array((0 until saltFactor).map(lit): _*)))
    .withColumn("salted_key", concat(col(joinKey), lit("_"), col("salt")))
  
  saltedLeft.join(saltedRight, "salted_key")
    .drop("salt", "salted_key")
}

// Utilisation du salted join
val saltedResult = saltedJoin(largeTable, smallTable, "user_id")

💡Gestion du Cache et de la Persistance

Stratégies de Cache Avancées

import org.apache.spark.storage.StorageLevel

// Différents niveaux de stockage
val df = spark.read.parquet("/data/input")

// Cache en mémoire uniquement
df.cache() // Équivalent à MEMORY_ONLY

// Cache avec sérialisation
df.persist(StorageLevel.MEMORY_ONLY_SER)

// Cache avec fallback sur disque
df.persist(StorageLevel.MEMORY_AND_DISK)

// Cache avec réplication
df.persist(StorageLevel.MEMORY_AND_DISK_2)

// Cache off-heap (nécessite Tachyon/Alluxio)
df.persist(StorageLevel.OFF_HEAP)

// Gestion intelligente du cache
class CacheManager(spark: SparkSession) {
  private val cachedDataFrames = scala.collection.mutable.Map[String, DataFrame]()
  
  def cacheWithEviction(name: String, df: DataFrame, storageLevel: StorageLevel = StorageLevel.MEMORY_AND_DISK): DataFrame = {
    // Vérifier l'utilisation mémoire
    val memoryUsage = getMemoryUsage()
    
    if (memoryUsage > 0.8) {
      // Éviction LRU
      evictLeastRecentlyUsed()
    }
    
    val cachedDF = df.persist(storageLevel)
    cachedDataFrames(name) = cachedDF
    cachedDF
  }
  
  def evictLeastRecentlyUsed(): Unit = {
    // Logique d'éviction basée sur l'utilisation
    cachedDataFrames.headOption.foreach { case (name, df) =>
      df.unpersist()
      cachedDataFrames.remove(name)
    }
  }
  
  private def getMemoryUsage(): Double = {
    val sc = spark.sparkContext
    val executorInfos = sc.getExecutorMemoryStatus
    val totalMemory = executorInfos.values.map(_._1).sum
    val usedMemory = executorInfos.values.map(x => x._1 - x._2).sum
    usedMemory.toDouble / totalMemory
  }
}

val cacheManager = new CacheManager(spark)
val cachedDF = cacheManager.cacheWithEviction("user_data", df)

Optimisation des Checkpoints

// Configuration des checkpoints
spark.sparkContext.setCheckpointDir("/data/checkpoints")

// Checkpoint intelligent basé sur la complexité
def smartCheckpoint(df: DataFrame, lineageDepth: Int = 10): DataFrame = {
  if (getLineageDepth(df) > lineageDepth) {
    df.checkpoint()
  } else {
    df
  }
}

def getLineageDepth(df: DataFrame): Int = {
  // Calculer la profondeur du lineage
  def countDependencies(rdd: org.apache.spark.rdd.RDD[_]): Int = {
    if (rdd.dependencies.isEmpty) 1
    else 1 + rdd.dependencies.map(dep => countDependencies(dep.rdd)).max
  }
  countDependencies(df.rdd)
}

// Utilisation
val processedDF = df
  .filter(col("status") === "active")
  .groupBy("category")
  .agg(sum("amount").as("total"))

val checkpointedDF = smartCheckpoint(processedDF)

💡Optimisation des Opérations I/O

Formats de Fichiers Optimaux

// Configuration Parquet optimisée
val optimizedParquetDF = spark.read
  .option("mergeSchema", "false")
  .option("filterPushdown", "true")
  .option("columnPruning", "true")
  .parquet("/data/input")

// Écriture Parquet optimisée
df.write
  .mode("overwrite")
  .option("compression", "snappy")
  .option("parquet.block.size", "134217728") // 128MB
  .option("parquet.page.size", "1048576")    // 1MB
  .option("parquet.dictionary.page.size", "1048576")
  .partitionBy("year", "month")
  .parquet("/data/output")

// Delta Lake pour les mises à jour fréquentes
import io.delta.tables._

// Écriture Delta
df.write
  .format("delta")
  .mode("overwrite")
  .option("overwriteSchema", "true")
  .save("/data/delta-table")

// Optimisation Delta
val deltaTable = DeltaTable.forPath(spark, "/data/delta-table")

// OPTIMIZE pour compacter les fichiers
deltaTable.optimize().executeCompaction()

// Z-ORDER pour optimiser les requêtes
deltaTable.optimize().executeZOrderBy("user_id", "timestamp")

// VACUUM pour nettoyer les anciens fichiers
deltaTable.vacuum(168) // Garder 7 jours d'historique

Optimisation des Lectures

// Lecture avec prédicats pushdown
val filteredDF = spark.read
  .parquet("/data/partitioned")
  .filter(col("year") === 2024 && col("month") >= 6)
  .select("user_id", "amount", "timestamp")

// Lecture avec colonnes projetées
val projectedDF = spark.read
  .option("columnPruning", "true")
  .parquet("/data/wide_table")
  .select("id", "name", "email") // Seulement les colonnes nécessaires

// Lecture streaming optimisée
val streamingDF = spark.readStream
  .format("kafka")
  .option("kafka.bootstrap.servers", "localhost:9092")
  .option("subscribe", "events")
  .option("maxOffsetsPerTrigger", "1000000")
  .option("startingOffsets", "latest")
  .load()

// Traitement micro-batch optimisé
val query = streamingDF
  .selectExpr("CAST(value AS STRING)")
  .writeStream
  .format("delta")
  .option("checkpointLocation", "/data/checkpoints/streaming")
  .option("mergeSchema", "true")
  .trigger(Trigger.ProcessingTime("30 seconds"))
  .start("/data/streaming-output")

💡Optimisation SQL et Catalyst

Utilisation du Catalyst Optimizer

// Activation des optimisations adaptatives (Spark 3.0+)
spark.conf.set("spark.sql.adaptive.enabled", "true")
spark.conf.set("spark.sql.adaptive.coalescePartitions.enabled", "true")
spark.conf.set("spark.sql.adaptive.skewJoin.enabled", "true")
spark.conf.set("spark.sql.adaptive.localShuffleReader.enabled", "true")

// Optimisation des jointures
spark.conf.set("spark.sql.adaptive.advisoryPartitionSizeInBytes", "134217728") // 128MB
spark.conf.set("spark.sql.adaptive.coalescePartitions.minPartitionNum", "1")
spark.conf.set("spark.sql.adaptive.skewJoin.skewedPartitionThresholdInBytes", "268435456") // 256MB

// Analyse des statistiques pour l'optimiseur
spark.sql("ANALYZE TABLE my_table COMPUTE STATISTICS")
spark.sql("ANALYZE TABLE my_table COMPUTE STATISTICS FOR COLUMNS user_id, amount")

// Hints pour guider l'optimiseur
val optimizedQuery = spark.sql("""
  SELECT /*+ BROADCAST(small_table) */ 
    large_table.user_id,
    large_table.amount,
    small_table.category
  FROM large_table
  JOIN small_table ON large_table.category_id = small_table.id
  WHERE large_table.amount > 1000
""")

// Utilisation des CTE pour la réutilisation
val complexQuery = spark.sql("""
  WITH user_stats AS (
    SELECT user_id, 
           COUNT(*) as transaction_count,
           SUM(amount) as total_amount
    FROM transactions 
    WHERE date >= '2024-01-01'
    GROUP BY user_id
  ),
  high_value_users AS (
    SELECT user_id 
    FROM user_stats 
    WHERE total_amount > 10000
  )
  SELECT u.*, s.transaction_count, s.total_amount
  FROM users u
  JOIN high_value_users h ON u.user_id = h.user_id
  JOIN user_stats s ON u.user_id = s.user_id
""")

Optimisation des UDF

// UDF Scala optimisée
import org.apache.spark.sql.expressions.UserDefinedFunction
import org.apache.spark.sql.functions.udf

// UDF simple
val calculateScore: UserDefinedFunction = udf((amount: Double, category: String) => {
  category match {
    case "premium" => amount * 1.5
    case "standard" => amount * 1.2
    case _ => amount
  }
})

// UDF vectorisée (Pandas UDF pour Python)
// Équivalent Scala avec des opérations vectorisées
def calculateScoreVectorized(df: DataFrame): DataFrame = {
  df.withColumn("score", 
    when(col("category") === "premium", col("amount") * 1.5)
    .when(col("category") === "standard", col("amount") * 1.2)
    .otherwise(col("amount"))
  )
}

// Comparaison de performance
val dfWithUDF = df.withColumn("score", calculateScore(col("amount"), col("category")))
val dfVectorized = calculateScoreVectorized(df)

// Préférer les opérations natives Spark
val nativeOperations = df
  .withColumn("score", 
    when(col("category") === "premium", col("amount") * 1.5)
    .when(col("category") === "standard", col("amount") * 1.2)
    .otherwise(col("amount"))
  )

💡Monitoring et Debugging

Métriques de Performance

// Classe utilitaire pour le monitoring
class SparkPerformanceMonitor(spark: SparkSession) {
  
  def getExecutorMetrics(): Map[String, Any] = {
    val sc = spark.sparkContext
    val executorInfos = sc.getExecutorInfos
    
    Map(
      "activeExecutors" -> executorInfos.size,
      "totalCores" -> executorInfos.map(_.totalCores).sum,
      "maxMemory" -> executorInfos.map(_.maxMemory).sum,
      "memoryUsed" -> executorInfos.map(_.memoryUsed).sum
    )
  }
  
  def getJobMetrics(jobId: Int): Option[Map[String, Any]] = {
    val sc = spark.sparkContext
    sc.statusTracker.getJobInfo(jobId).map { jobInfo =>
      Map(
        "jobId" -> jobInfo.jobId,
        "status" -> jobInfo.status,
        "numTasks" -> jobInfo.numTasks,
        "numActiveTasks" -> jobInfo.numActiveTasks,
        "numCompletedTasks" -> jobInfo.numCompletedTasks,
        "numFailedTasks" -> jobInfo.numFailedTasks
      )
    }
  }
  
  def analyzeDataSkew(df: DataFrame, partitionCol: String): Unit = {
    val partitionCounts = df.groupBy(col(partitionCol))
      .count()
      .collect()
      .map(_.getLong(1))
    
    val mean = partitionCounts.sum.toDouble / partitionCounts.length
    val variance = partitionCounts.map(count => math.pow(count - mean, 2)).sum / partitionCounts.length
    val stdDev = math.sqrt(variance)
    val coefficientOfVariation = stdDev / mean
    
    println(s"Partition analysis for column '$partitionCol':")
    println(s"Mean partition size: ${mean.toLong}")
    println(s"Standard deviation: ${stdDev.toLong}")
    println(s"Coefficient of variation: $coefficientOfVariation")
    
    if (coefficientOfVariation > 0.5) {
      println("WARNING: High data skew detected!")
    }
  }
}

val monitor = new SparkPerformanceMonitor(spark)
println(monitor.getExecutorMetrics())
monitor.analyzeDataSkew(df, "user_id")

Profiling et Debugging

// Configuration pour le profiling détaillé
spark.conf.set("spark.sql.execution.arrow.pyspark.enabled", "true")
spark.conf.set("spark.sql.execution.arrow.pyspark.fallback.enabled", "true")
spark.conf.set("spark.sql.adaptive.logLevel", "INFO")

// Fonction de profiling
def profileOperation[T](operation: => T, operationName: String): T = {
  val startTime = System.currentTimeMillis()
  val result = operation
  val endTime = System.currentTimeMillis()
  
  println(s"Operation '$operationName' took ${endTime - startTime}ms")
  
  // Log des métriques Spark
  val sc = spark.sparkContext
  val lastStageId = sc.statusTracker.getJobIds().lastOption
    .flatMap(jobId => sc.statusTracker.getJobInfo(jobId))
    .map(_.stageIds.max)
  
  lastStageId.foreach { stageId =>
    sc.statusTracker.getStageInfo(stageId).foreach { stageInfo =>
      println(s"Stage $stageId: ${stageInfo.numCompletedTasks}/${stageInfo.numTasks} tasks completed")
      println(s"Stage duration: ${stageInfo.submissionTime.map(System.currentTimeMillis() - _).getOrElse(0)}ms")
    }
  }
  
  result
}

// Utilisation
val result = profileOperation({
  df.groupBy("category")
    .agg(sum("amount").as("total"))
    .collect()
}, "GroupBy and Sum operation")

💡Patterns d'Optimisation Avancés

Pipeline d'Optimisation Automatique

class SparkOptimizationPipeline(spark: SparkSession) {
  
  def optimizeDataFrame(df: DataFrame): DataFrame = {
    var optimizedDF = df
    
    // 1. Analyse et optimisation du partitioning
    optimizedDF = optimizePartitioning(optimizedDF)
    
    // 2. Optimisation du cache
    optimizedDF = optimizeCache(optimizedDF)
    
    // 3. Optimisation des jointures
    optimizedDF = optimizeJoins(optimizedDF)
    
    optimizedDF
  }
  
  private def optimizePartitioning(df: DataFrame): DataFrame = {
    val currentPartitions = df.rdd.getNumPartitions
    val recordCount = df.count()
    val optimalPartitions = calculateOptimalPartitions(recordCount)
    
    if (math.abs(currentPartitions - optimalPartitions) > optimalPartitions * 0.2) {
      println(s"Repartitioning from $currentPartitions to $optimalPartitions partitions")
      df.repartition(optimalPartitions)
    } else {
      df
    }
  }
  
  private def calculateOptimalPartitions(recordCount: Long): Int = {
    val targetRecordsPerPartition = 100000
    val minPartitions = 200
    val maxPartitions = 2000
    
    (recordCount / targetRecordsPerPartition).toInt
      .max(minPartitions)
      .min(maxPartitions)
  }
  
  private def optimizeCache(df: DataFrame): DataFrame = {
    // Logique pour déterminer si le cache est bénéfique
    val lineageDepth = getLineageDepth(df)
    val estimatedSize = estimateDataFrameSize(df)
    
    if (lineageDepth > 5 && estimatedSize < getAvailableMemory() * 0.3) {
      println("Caching DataFrame based on lineage depth and size")
      df.cache()
    } else {
      df
    }
  }
  
  private def optimizeJoins(df: DataFrame): DataFrame = {
    // Analyse des jointures et optimisation automatique
    // Cette méthode nécessiterait une analyse plus complexe du plan d'exécution
    df
  }
  
  private def getLineageDepth(df: DataFrame): Int = {
    // Implémentation simplifiée
    5 // Placeholder
  }
  
  private def estimateDataFrameSize(df: DataFrame): Long = {
    // Estimation basée sur les statistiques
    df.queryExecution.optimizedPlan.stats.sizeInBytes
  }
  
  private def getAvailableMemory(): Long = {
    val executorMemory = spark.conf.get("spark.executor.memory", "1g")
    val memoryBytes = executorMemory.toLowerCase match {
      case mem if mem.endsWith("g") => mem.dropRight(1).toLong * 1024 * 1024 * 1024
      case mem if mem.endsWith("m") => mem.dropRight(1).toLong * 1024 * 1024
      case mem => mem.toLong
    }
    memoryBytes
  }
}

// Utilisation du pipeline d'optimisation
val optimizer = new SparkOptimizationPipeline(spark)
val optimizedDF = optimizer.optimizeDataFrame(df)

💡Configuration de Production

Configuration Complète pour la Production

#!/bin/bash
# spark-production-config.sh

# Configuration Spark optimisée pour la production
spark-submit \
  --class com.example.SparkApplication \
  --master yarn \
  --deploy-mode cluster \
  --driver-memory 8g \
  --driver-cores 4 \
  --executor-memory 28g \
  --executor-cores 5 \
  --num-executors 20 \
  --conf "spark.executor.memoryFraction=0.8" \
  --conf "spark.storage.memoryFraction=0.3" \
  --conf "spark.shuffle.memoryFraction=0.2" \
  --conf "spark.serializer=org.apache.spark.serializer.KryoSerializer" \
  --conf "spark.kryo.registrationRequired=false" \
  --conf "spark.kryo.unsafe=true" \
  --conf "spark.kryoserializer.buffer.max=1024m" \
  --conf "spark.sql.adaptive.enabled=true" \
  --conf "spark.sql.adaptive.coalescePartitions.enabled=true" \
  --conf "spark.sql.adaptive.skewJoin.enabled=true" \
  --conf "spark.sql.adaptive.localShuffleReader.enabled=true" \
  --conf "spark.sql.adaptive.advisoryPartitionSizeInBytes=134217728" \
  --conf "spark.shuffle.compress=true" \
  --conf "spark.shuffle.spill.compress=true" \
  --conf "spark.rdd.compress=true" \
  --conf "spark.broadcast.compress=true" \
  --conf "spark.io.compression.codec=snappy" \
  --conf "spark.network.timeout=800s" \
  --conf "spark.executor.heartbeatInterval=60s" \
  --conf "spark.dynamicAllocation.enabled=true" \
  --conf "spark.dynamicAllocation.minExecutors=5" \
  --conf "spark.dynamicAllocation.maxExecutors=50" \
  --conf "spark.dynamicAllocation.initialExecutors=10" \
  --conf "spark.executor.extraJavaOptions=-XX:+UseG1GC -XX:+UnlockDiagnosticVMOptions -XX:+G1PrintRegionRememberSetInfo -XX:+PrintGC -XX:+PrintGCDetails -XX:+PrintGCTimeStamps -Xloggc:/var/log/spark/gc-executor.log" \
  --conf "spark.driver.extraJavaOptions=-XX:+UseG1GC -XX:+UnlockDiagnosticVMOptions -XX:+G1PrintRegionRememberSetInfo -XX:+PrintGC -XX:+PrintGCDetails -XX:+PrintGCTimeStamps -Xloggc:/var/log/spark/gc-driver.log" \
  --conf "spark.eventLog.enabled=true" \
  --conf "spark.eventLog.dir=hdfs://namenode:8020/var/log/spark-events" \
  --conf "spark.history.fs.logDirectory=hdfs://namenode:8020/var/log/spark-events" \
  application.jar

Monitoring et Alerting en Production

// Configuration de monitoring pour la production
object ProductionMonitoring {
  
  def setupMetrics(spark: SparkSession): Unit = {
    // Configuration des métriques
    spark.conf.set("spark.metrics.conf.*.sink.graphite.class", "org.apache.spark.metrics.sink.GraphiteSink")
    spark.conf.set("spark.metrics.conf.*.sink.graphite.host", "graphite-server")
    spark.conf.set("spark.metrics.conf.*.sink.graphite.port", "2003")
    spark.conf.set("spark.metrics.conf.*.sink.graphite.period", "10")
    spark.conf.set("spark.metrics.conf.*.sink.graphite.unit", "seconds")
    
    // Listener personnalisé pour les alertes
    spark.sparkContext.addSparkListener(new ProductionSparkListener())
  }
  
  class ProductionSparkListener extends SparkListener {
    override def onJobEnd(jobEnd: SparkListenerJobEnd): Unit = {
      val duration = jobEnd.time - jobEnd.jobResult.jobId
      
      if (duration > 300000) { // Plus de 5 minutes
        sendAlert(s"Long running job detected: Job ${jobEnd.jobId} took ${duration}ms")
      }
    }
    
    override def onTaskEnd(taskEnd: SparkListenerTaskEnd): Unit = {
      val metrics = taskEnd.taskMetrics
      val gcTime = metrics.jvmGCTime
      val taskDuration = taskEnd.taskInfo.duration
      
      if (gcTime > taskDuration * 0.1) { // GC > 10% du temps de tâche
        sendAlert(s"High GC time detected: ${gcTime}ms out of ${taskDuration}ms")
      }
      
      if (metrics.shuffleReadMetrics.fetchWaitTime > taskDuration * 0.3) {
        sendAlert(s"High shuffle wait time: ${metrics.shuffleReadMetrics.fetchWaitTime}ms")
      }
    }
    
    private def sendAlert(message: String): Unit = {
      // Intégration avec votre système d'alerting
      println(s"ALERT: $message")
      // Webhook, email, Slack, etc.
    }
  }
}

💡Conclusion

L'optimisation d'Apache Spark nécessite une approche holistique qui combine :

Configuration Appropriée

Optimisation des Données

Optimisation du Code

Monitoring Continu

Les techniques présentées dans cet article vous permettront d'optimiser significativement les performances de vos applications Spark en production. N'oubliez pas que l'optimisation est un processus itératif qui nécessite des tests et des ajustements constants.

Pour un accompagnement dans l'optimisation de vos pipelines Spark, contactez-moi pour une consultation personnalisée.

À propos de l'auteur

Florian Courouge - Expert DevOps et Apache Kafka avec plus de 5 ans d'expérience dans l'architecture de systèmes distribués et l'automatisation d'infrastructures.

Cet article vous a été utile ?

Découvrez mes autres articles techniques ou contactez-moi pour discuter de vos projets DevOps et Kafka.