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.

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.