Diagnosing and Fixing Java Heap Space OOM During Spark Shuffles

The Problem: Executor Memory Exhaustion During Shuffle

An OutOfMemoryError: Java heap space during a Spark shuffle occurs when a single task attempts to process more data than the executor's allocated heap can hold. This typically happens during wide transformations—such as join, groupBy, or reduceByKey—where data is redistributed across the cluster.

The critical takeaway is that increasing total cluster memory rarely fixes this if the root cause is data skew. If one partition is 10x larger than others, that specific executor will crash regardless of how much memory you add to the rest of the cluster.

Quick Diagnostic Matrix

Symptom in Spark UI	Likely Cause	Primary Metric to Check
One or few tasks take hours; others finish in seconds	Data Skew	Max Shuffle Read vs Median Shuffle Read
All tasks fail with OOM at roughly the same time	Under-provisioned Heap	Shuffle Read Size per Task vs spark.executor.memory
Frequent "GC overhead limit exceeded" before OOM	Memory Pressure/Fragmentation	GC Time in Stage tab
OOM occurs immediately after a .collect()	Driver Memory Exhaustion	Driver Heap usage

Step-by-Step Diagnostic Workflow

1. Identify the Failure Pattern

Open the Spark UI and navigate to the Stages tab. Locate the failed stage and examine the Summary Metrics table.

• Check for Skew: Compare the "Max" shuffle read size to the "Median". If the Max is significantly higher (e.g., 5x or more), you have a data skew problem.
• Check for Under-provisioning: If the Median shuffle read size is close to your spark.executor.memory setting, your partitions are simply too large for your hardware.

2. Analyze Memory Distribution

Check the Storage tab. If a large portion of the heap is occupied by cached RDDs or DataFrames, there is less room for the Execution Memory required for shuffles. Spark divides the heap based on spark.memory.fraction (default 0.6). If storage is full, execution memory is squeezed.

3. Review Executor Logs

Access the logs via the Yarn Resource Manager or Kubernetes dashboard. Look for the specific exception trace. A java.lang.OutOfMemoryError: Java heap space during a shuffle usually points to the ShuffleMapStage or ResultStage.

Fixes Based on Findings

Scenario A: Data Skew Detected

If a few keys are disproportionately large, increasing memory is a temporary bandage. Instead, use salting. Salting adds a random prefix to the join key to spread the data across more partitions.

// Example: Salting a skewed key in PySpark
from pyspark.sql import functions as F
import random

# Add a random salt to the skewed DataFrame
df_skewed = df_skewed.withColumn("salt", (F.rand() * 10).cast("int"))
df_skewed = df_skewed.withColumn("salted_key", F.concat(F.col("key"), F.lit("_"), F.col("salt")))

# Explode the lookup table to match the salt range
df_lookup = df_lookup.withColumn("salt", F.explode(F.array([F.lit(i) for i in range(10)])))
df_lookup = df_lookup.withColumn("salted_key", F.concat(F.col("key"), F.lit("_"), F.col("salt")))

# Join on the salted key
result = df_skewed.join(df_lookup, "salted_key")

Scenario B: Uniformly Large Partitions

If data is evenly distributed but partitions are too large, increase the number of shuffle partitions. This reduces the amount of data each task must process in memory.

Action: Run this command in your Spark session or add it to spark-submit:

--conf spark.sql.shuffle.partitions=1000

Risk: Setting this too high (e.g., 10,000+) can create excessive small files and increase scheduling overhead.

Scenario C: Memory Misconfiguration

If you have ample physical RAM but still hit OOM, adjust the memory fraction. If you aren't using .persist() or .cache(), you can give more room to execution memory.

Action: Increase spark.memory.fraction (e.g., to 0.8) to allow more heap for shuffles and joins.

Verification and Limitations

To verify the fix, rerun the job and monitor the Stages tab. The "Max" shuffle read size should now be closer to the "Median," and the GC time should decrease.

Limitations:

• Increasing spark.executor.memory without decreasing spark.executor.cores can lead to longer Garbage Collection (GC) pauses, which may trigger ExecutorHeartbeat timeouts.
• These fixes address executor OOM. If the error occurs during .collect(), the issue is spark.driver.memory, not the shuffle configuration.

Rollback Procedure

If the changes cause instability (e.g., excessive small files or driver timeouts), revert the configurations to their defaults:

• Reset spark.sql.shuffle.partitions to 200.
• Reset spark.memory.fraction to 0.6.
• Remove salting logic from the codebase to return to standard join keys.