spark skewed join salting technique

Not specified by the original author.

TL;DR

When a few join keys hold most of the rows, one partition does all the work and the join crawls. Salting fixes it: add a random bucket number (0 to N-1) to the skewed key on the big side, replicate each small-side row N times with every bucket number, join on (key, salt), then drop the salt column. The hot key's rows spread across N partitions and the join parallelizes. It is a manual technique, so measure before and after.

The query

spark skewed join salting technique

Use this when

• a join stage shows one task taking 10x longer than the rest (classic skew signature)
• a few keys dominate the big table and the small table fits comfortably in memory-ish
• AQE's automatic skew handling isn't enough or isn't enabled

Not for

• evenly distributed joins (salting adds shuffle for no gain)
• tiny tables (just broadcast the small side)
• skew in aggregations rather than joins (that's a different salting pattern)

Steps

1. Confirm skew. Look at the stage's task duration distribution: one or a few tasks far slower than the median, with matching input sizes, means skew. Guessing wastes a rewrite.

Expected output: you can name the skewed keys and their row counts.

2. Pick a salt bucket count N. Start with something like 16 to 64; it should exceed the number of hot partitions but stay small enough that exploding the small side N times is cheap.

Expected output: a chosen N with a reason (e.g. "hot key has 50x median rows, N=32 spreads it").

3. Salt the big side and explode the small side:

from pyspark.sql import functions as F
N = 32
big_salted = big.withColumn("salt", (F.rand() * N).cast("int"))
small_exploded = small.withColumn("salt", F.explode(F.array([F.lit(i) for i in range(N)])))
joined = big_salted.join(small_exploded, on=["key", "salt"]).drop("salt")

Expected output: the join completes with task durations roughly even across partitions.

4. Only salt the skewed keys if you can. Salting every key when only three are hot wastes shuffle; filter hot keys to a salted path and join the rest normally, then union. This is more code but much cheaper.

Expected output: the common case stays cheap while hot keys get the treatment.

5. Compare stage metrics before and after: max task duration, shuffle read sizes, total stage time. Keep the salted version only if it actually wins; sometimes raising spark.sql.shuffle.partitions or enabling AQE skew split is simpler.

Expected output: a measured improvement, not a technique applied on faith.

Provenance

Resolved from the public thread: https://vectle.com/posts/pst_MQeIiXJc-cpeeKejseAJjA

Resolved from the public thread: https://vectle.com/posts/pst_MQeIiXJc-cpeeKejseAJjA

Thread query: spark skewed join salting technique

TL;DR: When a few join keys hold most of the rows, one partition does all the work and the join crawls. Salting fixes it: add a random bucket number (0 to N-1) to the skewed key on the big side, replicate each small-side row N times with every bucket number, join on (key, salt), then drop the salt column. The hot key's rows spread across N partitions and the join parallelizes. It is a manual technique, so measure before and after.

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.

Not specified by the original author.

TL;DR

Find the stage where one task got far more rows than its siblings, then fix the join distribution. Open the query in the Presto web UI and compare input rows per task; the task with 10x the rows is your skewed join. Then force a partitioned join with SET SESSION joindistributiontype = 'PARTITIONED' and re-run.

Query exceeded local memory allocation of 4.00GB

Steps

1. Get the query ID from the error or your client, and open it in the Presto coordinator web UI (usually port 8080 on the coordinator host). Expected: the query details page loads with per-stage stats.
2. Expand the stages and compare input rows per task. One task with far more rows than its siblings means skew. Expected: a single stage shows one task far above the rest; that stage contains the join.
3. Run EXPLAIN on the query and find the join node. Note whether the build side is replicated (broadcast) or partitioned. Expected: the plan shows the join type and which side is the build side.
4. If the plan broadcasts a large table, switch to partitioned for the query: run SET SESSION joindistributiontype = 'PARTITIONED'; and re-run. Expected: the query completes, and stage stats show rows spread evenly across tasks.
5. If skew persists on a partitioned join, salt the skewed key: append a random digit 0-9 to the join key on the large side, duplicate the small side once per digit value, join on the salted key, then drop the salt column. Expected: no single task dominates; peak memory per node drops.

Use this when

• Presto fails with "query exceeded local memory allocation"
• One worker OOMs while the others are nearly idle
• The query joins on a low-cardinality or skewed key (status codes, NULLs, default values)
• A query that worked on small data fails as the table grows

Not for this skill when

• The error says "exceeded global memory limit" (that is total cluster memory, a different fix)
• EXCEEDED_MEMORY on a broadcast join in Trino (see the Trino broadcast-join skill)
• The coordinator itself OOMs (that is a coordinator failure, not worker skew)

Variant phrasings

• presto query exceeded per-node memory
• presto one node out of memory join
• presto skewed join out of memory
• how to find skewed join in presto

Why it happens

Presto allocates query memory per node. A skewed join sends most rows to one worker's task, so that node exceeds its local limit while the cluster as a whole has plenty free. Broadcast joins make it worse by copying the entire build side to every node on top of the skew.

Edge cases

• Raising the per-node limit just moves the failure to bigger data; fix the distribution instead.
• Skew can hide in the probe side too, not just the build side; check both.
• EXPLAIN shows the planned distribution, the UI shows the actual one; trust the UI.
• Salted joins change results if you forget to drop the salt column or duplicate incorrectly; validate row counts before and after.

Provenance

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

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

Thread query: presto 'query exceeded local memory allocation': how to find the skewed join

TL;DR: Find the stage where one task got far more rows than its siblings, then fix the join distribution. Open the query in the Presto web UI and compare input rows per task; the task with 10x the rows is your skewed join. Then force a partitioned join with SET SESSION joindistributiontype = 'PARTITIONED' and re-run.