Effective Spark Performance Tuning and Troubleshooting Guide

Spark Performance Optimization

Typical production Spark submit script:

/usr/local/spark/bin/spark-submit \
  --class com.atguigu.spark.WordCount \
  --num-executors 80 \
  --driver-memory 6g \
  --executor-memory 6g \
  --executor-cores 3 \
  --master yarn \
  --deploy-mode cluster \
  --queue root.default \
  --conf spark.yarn.executor.memoryOverhead=2048 \
  --conf spark.core.connection.ack.wait.timeout=300 \
  /usr/local/spark/spark.jar

Key optimizations:

Reuse RDDs to avoid duplicate computation.

Persist frequently used RDDs to memory or disk.

Apply filters as early as possible.

Parallelism Tuning

Set the number of tasks to 2–3 times the total CPU cores of the Spark job.

val conf = new SparkConf().set("spark.default.parallelism", "500")

Broadcast Large Variables

Broadcast variables keep a single copy per executor, reducing memory consumption compared with normal task‑level copies.

Kryo Serialization

Spark uses Java serialization by default; Kryo can be up to ten times faster but requires registration of custom classes. Since Spark 2.0.0, simple types are serialized with Kryo automatically.

public class MyKryoRegistrator implements KryoRegistrator {
  @Override
  public void registerClasses(Kryo kryo) {
    kryo.register(StartupReportLogs.class);
  }
}
// SparkConf setup
val conf = new SparkConf()
  .setMaster(...).setAppName(...)
  .set("spark.serializer", "org.apache.spark.serializer.KryoSerializer")
  .set("spark.kryo.registrator", "atguigu.com.MyKryoRegistrator")

Locality Wait Time

Default wait is 3 seconds; increasing it can improve task locality but must be balanced to avoid excessive job duration.

val conf = new SparkConf().set("spark.locality.wait", "6")

Operator Tuning

Use mapPartitions instead of map for JDBC writes to reduce connection overhead, and be aware of possible OOM when processing very large partitions.

Similarly, foreachPartition can optimize database writes, but large partitions may still cause OOM.

Combine filter with coalesce to balance partition sizes after filtering.

Use repartition to increase Spark SQL parallelism because Spark SQL ignores spark.default.parallelism settings. reduceByKey performs map‑side aggregation, reducing shuffle I/O, disk usage, and memory consumption compared with groupByKey.

JVM Tuning

Both full GC and minor GC pause all JVM threads (stop‑the‑world). Reduce the memory fraction used for storage to free more memory for execution:

val conf = new SparkConf().set("spark.storage.memoryFraction", "0.4")

Unified memory management automatically balances storage and execution; manual tuning is rarely needed.

Increase executor off‑heap memory when processing billions of records:

# In spark-submit
--conf spark.yarn.executor.memoryOverhead=2048

Adjust connection timeout to avoid shuffle failures caused by network latency:

# In spark-submit
--conf spark.core.connection.ack.wait.timeout=300

Spark Data Skew Solutions

Symptoms include a few very slow tasks or OOM errors on specific tasks.

Diagnosis steps:

Inspect shuffle operators (e.g., reduceByKey, groupByKey, join) for potential skew.

Check Spark logs for stage and task details.

Shuffle Tuning

Increase map‑side buffer size ( spark.shuffle.file.buffer) to reduce disk spills.

Increase reduce‑side fetch buffer ( spark.reducer.maxSizeInFlight).

Raise retry count ( spark.shuffle.io.maxRetries) and retry wait ( spark.shuffle.io.retryWait).

Adjust SortShuffle bypass threshold ( spark.shuffle.sort.bypassMergeThreshold) to skip sorting when possible.

val conf = new SparkConf()
  .set("spark.shuffle.file.buffer", "64")
  .set("spark.reducer.maxSizeInFlight", "96")
  .set("spark.shuffle.io.maxRetries", "6")
  .set("spark.shuffle.io.retryWait", "60s")
  .set("spark.shuffle.sort.bypassMergeThreshold", "400")

Avoid Shuffle

Pre‑aggregate data in Hive before Spark, or change key granularity to reduce skew.

Filter out problematic keys (e.g., null values) before shuffle.

Increase reduce‑side parallelism via spark.sql.shuffle.partitions or by passing a parallelism argument to reduceByKey.

Random Key Double Aggregation

Prefix keys with a random number, perform a first aggregation, then remove the prefix and aggregate again. This spreads a hot key across many tasks.

Applicable to groupByKey and reduceByKey but not to joins.

Map Join

When one side of a join is small, broadcast it and use a map‑side join to eliminate shuffle.

Note: RDDs cannot be broadcast directly; collect data to the driver first.

Sample‑Based Join for Skewed Keys

Sample a fraction of data, identify skewed keys, extract those keys into a separate RDD, and join them separately so Spark distributes the data across many reduce tasks.

Random Key Expansion Join

Expand one RDD by duplicating each record with multiple prefixed keys, and map the other RDD with random prefixes to dilute the hot key effect. This approach is limited when both RDDs are large.

Spark Troubleshooting

Reduce‑side Buffer OOM : Large map‑side writes can fill the default 48 MB reduce buffer, causing OOM. Reducing the buffer (e.g., to 12 MB) trades performance for stability.

Shuffle File Not Found due to GC : GC pauses stop BlockManager and Netty, leading to temporary shuffle file errors. Increase retry count and wait time:

val conf = new SparkConf()
  .set("spark.shuffle.io.maxRetries", "60")
  .set("spark.shuffle.io.retryWait", "60s")

Serialization Errors : Ensure custom classes used in RDDs and closures implement java.io.Serializable. Avoid non‑serializable types such as java.sql.Connection inside RDD elements or functions.

Custom RDD element classes must be serializable.

External variables captured by closures must be serializable.

Do not use third‑party non‑serializable types inside RDDs.

Null Return in Operators : Return a sentinel value (e.g., -1) instead of null, then filter it out and optionally coalesce the result.

YARN‑Client vs YARN‑Cluster Network Load : In client mode the driver runs locally, causing high network traffic to many executors. Use cluster mode for production to avoid this issue.

PermSize for Driver : YARN‑cluster mode may use a default PermGen of 82 MB, causing failures for complex SparkSQL queries. Increase PermSize via:

# Set driver PermGen size (default 128 MB, max 256 MB)
--conf spark.driver.extraJavaOptions="-XX:PermSize=128M -XX:MaxPermSize=256M"

JVM Stack Overflow in SparkSQL : Deep recursion from many OR clauses can overflow the stack. Split complex SQL into multiple statements, each with fewer than ~100 predicates.

Persistence and Checkpoint

Cache an RDD for fast reuse; checkpoint provides a reliable fallback if the cache is lost, at the cost of writing data to HDFS.

Source: https://blog.csdn.net/qq_42180284/article/details/103945403