spark executor "OutOfMemoryError: Java heap space" partition tuning

TL;DR

An executor got a partition bigger than its heap. Add more partitions so each one is smaller, fix skewed keys that pile data onto one executor, and give executors more memory if the data is legitimately large. Check partition sizes first; guessing at memory settings wastes the tuning cycle.

Error

OutOfMemoryError: Java heap space

(on the executor, typically during a shuffle, join, or aggregation)

Steps

1. Confirm the failure is partition size, not a leak. Look at the stage in the Spark UI: task sizes and shuffle read sizes per task.

Expected output: a few tasks with huge shuffle reads while most are small. That pattern is oversized or skewed partitions. Uniformly large tasks point at genuinely big data instead.

2. Raise the shuffle partition count so each partition shrinks:

spark.conf.set("spark.sql.shuffle.partitions", 400)

Expected output: the same shuffle now spreads across 400 tasks. Start from 2-4x the current value; the default 200 is too low for large shuffles.

3. If one key dominates (skew), salt it before the wide operation:

from pyspark.sql import functions as F
salted = df.withColumn("salt", (F.rand() * 10).cast("int"))

Expected output: the hot key's rows spread across 10 salt buckets. Aggregate per salted key, then aggregate again without the salt to get final results.

4. Repartition narrow, skewed inputs before the join or aggregation:

df = df.repartition(400, "join_key")

Expected output: input partitions are even before the expensive step, so no single executor starts overloaded.

5. If partitions are already small and it still OOMs, raise executor memory:

# spark-submit --conf spark.executor.memory=8g

Expected output: executors survive the stage. Only do this after steps 2-4; memory without partitioning just moves the crash to bigger data.

When to use

• Executors fail with Java heap space during shuffles, joins, or group-bys
• The Spark UI shows a few giant tasks among small ones
• The job worked on smaller data and broke as data grew

When not to use

• The driver OOMs (that is driver memory, or a collect bringing data home)
• The error is "Container killed by YARN" (that is off-heap or container limits)
• A single row is larger than the heap (no partition count fixes that; fix the row)

Variant phrasings

OOM only on one stage

That stage has the wide operation. Tune the partitions feeding that stage, not the whole job.

OOM after adding a broadcast join hint

The broadcast table is bigger than you thought. Check its size; remove the hint or raise the broadcast threshold deliberately.

Why it happens

Spark processes one partition per task in executor heap. Too few partitions, or one hot key, puts gigabytes into a single task's memory, and the JVM dies when the partition plus working overhead exceeds the heap. Partition count and skew control per-task memory; executor memory only sets the ceiling.

Edge cases

• coalesce after a big shuffle can re-create giant partitions; repartition instead when growing partition counts.
• spark.default.parallelism governs RDD shuffles while spark.sql.shuffle.partitions governs DataFrame shuffles; set the one your code uses.
• Over-partitioning has a cost: tens of thousands of tiny tasks add scheduling overhead, so stop when tasks run a few seconds each.

Provenance

Resolved from the public thread: https://vectle.com/posts/pstrkQvhc7UGoQ4xtBF2EUwQ