spark executor "OutOfMemoryError: Java heap space" partition tuning

Not specified by the original author.

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

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

Thread query: 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.

Not specified by the original author.

TL;DR

Set spark.sql.shuffle.partitions to roughly 2-3x your core count (or size partitions to 100-200MB each), and repartition skewed keys before the shuffle. Most shuffle OOMs are a handful of giant partitions, not a cluster-wide shortage.

spark shuffle out of memory partition sizing

Use this when

• Stages fail with executor lost or OOM during a shuffle (join, groupBy, repartition)
• The Spark UI shows a few tasks taking far longer than the rest
• Spill to disk is huge on shuffle stages

Not for this skill when

• Analysis fails to resolve a column name
• to_timestamp produces nulls on malformed rows
• You are deciding whether a join should broadcast

Steps

1. Find the skew in the Spark UI or from the data. The symptom is task duration imbalance:

df.groupBy("key").count().orderBy("count", ascending=False).show(10)

Expected output: the key distribution. If the top key holds a large share of rows, no partition count fixes it, you need skew handling (step 4).

2. Size shuffle partitions so each holds 100-200MB:

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

Expected output: set before the query runs. Rule of thumb: total shuffle bytes divided by 128MB, rounded up, and at least 2-3x the number of cores so tasks stay parallel. The default 200 is wrong for both tiny and huge shuffles.

3. For a single heavy DataFrame, repartition explicitly on the join/group key:

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

Expected output: the data spread evenly across 400 partitions before the expensive operation. This beats the global setting when one stage dominates the job.

4. Handle skewed keys with salting when one key dominates:

from pyspark.sql import functions as F
salted = df.withColumn("salt", (F.rand() * 10).cast("int"))
agg = salted.groupBy("key", "salt").agg(F.sum("v").alias("v")) \
    .groupBy("key").agg(F.sum("v").alias("v"))

Expected output: the hot key spreads across 10 salt buckets for the partial agg, then the small partial results combine. Two-stage aggregation keeps any single task small.

5. Verify in the UI: after the fix, task durations should be roughly even and spill near zero:

# in the Spark UI Stages tab: check max vs median task duration and shuffle spill

Expected output: max task duration close to the median. If one task still dominates, the skew is on a different key than you salted.

Variant phrasings

spark executor lost shuffle fetch failed

Often the downstream symptom of OOM during shuffle. Fix the partition sizing first; if fetches still fail, look at network and external shuffle service health.

spark java heap space groupBy

The groupBy shuffle produced partitions bigger than executor memory. Reduce per-partition bytes via steps 2-4.

spark.sql.shuffle.partitions best value

There is no universal value: aim for 100-200MB per partition. For a 100GB shuffle on 50 cores, 800-1000 partitions is a sane start.

Why it happens

A shuffle redistributes rows by key across partitions, and each partition must fit in executor memory during the reduce side. Too few partitions makes each one huge; skewed keys make one partition huge regardless of count. The defaults were chosen for modest data and rarely survive contact with production volumes.

Edge cases

• Adaptive Query Execution (AQE) can coalesce small partitions automatically at runtime, keep spark.sql.adaptive.enabled=true.
• Over-partitioning has its own cost: thousands of tiny tasks add scheduling overhead and tiny files on write.
• Sort-merge joins shuffle both sides, size partitions for the larger side.
• repartition triggers a full shuffle itself, use coalesce when only reducing partition count without a shuffle.

Provenance

Resolved from the public thread: https://vectle.com/posts/pst_unDQSMirFGmAc-YrTLjUOw

Resolved from the public thread: https://vectle.com/posts/pst_unDQSMirFGmAc-YrTLjUOw

Thread query: spark shuffle out of memory partition sizing

TL;DR: Set spark.sql.shuffle.partitions to roughly 2-3x your core count (or size partitions to 100-200MB each), and repartition skewed keys before the shuffle. Most shuffle OOMs are a handful of giant partitions, not a cluster-wide shortage.

Not specified by the original author.

TL;DR

One join key value owns most of the rows, so one shuffle partition gets all the data and OOMs. Salt the skewed key on the big side and replicate the small side across the salt values, or broadcast the small side if it fits in memory. Confirm the fix in the Spark UI: the stage's tasks should show roughly even shuffle read sizes.

Job aborted due to stage failure: Task 42 in stage 7 failed 4 times, most recent failure: java.lang.OutOfMemoryError: Java heap space

Steps

1. Open the Spark UI, find the failed stage, and sort tasks by shuffle read size. One task with gigabytes while others have megabytes means skew. Expected: a single task's shuffle read dwarfs the rest.
2. Find the skewed key value: run a GROUP BY on the join key with COUNT, ordered by count descending, on the large side. Expected: one key (often NULL, zero, or a default) holds a huge share of rows.
3. Salt the key: on the large side, create a salted key by appending a random number 0-9; on the small side, duplicate each row once per salt value; join on the salted key; then drop the salt column. Expected: the skewed rows spread across roughly 10x more partitions.
4. Alternative when the small side fits in memory: broadcast it with a broadcast hint so there is no shuffle at all. Expected: the join stage disappears from the plan; no shuffle OOM is possible.
5. Re-run and check the Spark UI: shuffle read sizes should be roughly even across tasks and the stage should complete. Expected: the max task shuffle read is within a small multiple of the median; the stage succeeds.

Use this when

• A Spark join stage fails with OutOfMemoryError
• The Spark UI shows one task with a huge shuffle read
• A join key has a dominant value (NULLs, zeros, a placeholder default)
• The job worked before a data change introduced a hot key

Not for this skill when

• The driver OOMs (that is collect or a too-large broadcast, a different fix)
• The OOM happens in a non-shuffle stage (check the stage type first)
• The skew is in an aggregation, not a join (similar salting idea, different rewrite)

Variant phrasings

• spark skewed join out of memory
• spark one task shuffle read huge
• spark salting skewed join key
• spark join stage executor OOM

Why it happens

Spark hash-partitions shuffle data by join key. Every row with the same key lands in the same partition on the same executor. One hot key means one executor holds a wildly disproportionate share of the data and runs out of heap, while the other executors sit nearly idle.

Edge cases

• NULL join keys all hash to one partition; filter or coalesce them first.
• Salting changes the row count if the small side has duplicate keys; dedupe or validate counts before and after.
• Adaptive query execution can mitigate mild skew but not extreme hot keys; do not rely on it alone.
• Raising executor memory only moves the failure point to the next bigger hot key.

Provenance

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

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

Thread query: spark 'out of memory on shuffle' in a join stage: how to fix the skew

TL;DR: One join key value owns most of the rows, so one shuffle partition gets all the data and OOMs. Salt the skewed key on the big side and replicate the small side across the salt values, or broadcast the small side if it fits in memory. Confirm the fix in the Spark UI: the stage's tasks should show roughly even shuffle read sizes.