Spark Data Skew

The symptom is a stage that sits at 199 of 200 tasks for most of its run. Open that stage in the Spark UI. The stage detail page shows summary metrics for all tasks, including duration and shuffle read size with records, and a table with the same figures per task. In an even stage the maximum is close to the median. In a skewed stage the maximum is many times the median, and the slow task is the one with the largest shuffle read.

Adaptive query execution uses the same comparison. Its own definition of a skewed partition is one larger than 5 times the median partition size and larger than 256 MB. In the example the median is 30 MB, so 5 times it is 150 MB, and the 4,000 MB partition clears both bars. A good field check is to apply that rule by eye: more than 5 times the median is worth a look.

To name the key, count rows per join key on each side and sort descending. One key far above the rest is the culprit. Check both sides, because the fix differs when the hot key is also large in the table you hoped to broadcast.

Adaptive query execution (AQE) re-plans a query from the real sizes of the shuffle output. Its skew join optimization splits, and replicates if needed, the oversized partitions of a sort-merge join into roughly even tasks. It runs when both spark.sql.adaptive.enabled and spark.sql.adaptive.skewJoin.enabled are true. The Spark documentation lists AQE as on by default since 3.2.0 and the skew setting as true since 3.0.0, so on a current release a skewed join may already be handled.

In the example, AQE sees one partition of 4,000 MB against a 30 MB median, and splits it toward the 64 MB advisory size. That's about 60 tasks. Each task reads its slice of the big side and the same matching partition of the small side, which is the replication the documentation mentions.

The limits matter more than the defaults. AQE changes join shuffles, so a skewed group by or window function is not covered. It can only split a side whose rows are safe to repeat: the preserved side of an outer join cannot be split, because every copy would emit its unmatched rows again, and a full outer join can split neither side. If splitting would force an extra shuffle somewhere else in the plan, the rewrite may be dropped. Read the final plan in the SQL tab of the Spark UI to see whether the join was split, instead of assuming it was.

If AQE fires but the stage is still slow, the usual cause is a hot key too large for the splitting to help, or a hot key present on both sides, where the join output for that key is the product of its row counts and no split makes that product small.

A broadcast hash join copies the small table to every executor and joins each big-table row where it already sits, so the big table is never shuffled and its hot key has no partition to pile into. Spark broadcasts on its own below spark.sql.autoBroadcastJoinThreshold, which the Spark tuning guide gives as 10485760 bytes, 10 MB, and -1 turns it off. A BROADCAST hint forces the choice even above the threshold.

In the example, customers has 2,000,000 rows. At 100 bytes a row that is 200 MB, well above the 10 MB default but small for an executor, so a hint makes the skew disappear. The hint is a request: the same guide says Spark doesn't guarantee it will use the strategy named, since not every strategy supports every join type.

The limit is memory. The table is built on the driver and held by every executor, so a broadcast of something that does not fit fails the job with an out of memory error.

Salting gives the hot key a second part. On the big side, each row of a hot key gets a random bucket number from 0 to N minus 1. On the small side, each row of a hot key is copied N times, once per bucket. The join then runs on the key and the bucket together, so the hot key's rows hash to as many partitions as it has buckets, and every big row still finds exactly one matching small row.

Choose N from the sizes. The hot key holds 4,000 MB and a comfortable task is the 64 MB advisory size, so 4,000 / 64 is about 63, and 64 is the round number. Each bucket then carries 625,000 rows, 62.5 MB. The 64 sub-keys hash independently across the 200 shuffle partitions, so some can share a partition and the spread is close to 64, not exactly.

Salt only the keys that are hot. Salting every key multiplies the small side by N: 2,000,000 customer rows become 128,000,000, more than the 100,000,000 events they're joined to. Salting only the hot key leaves 2,000,000 rows and adds 63 copies of the one hot row. The price is that you must know the hot keys, and a new hot key next month needs a new entry in the list.

Salting leaves the total work of the join the same and spreads it across more tasks. If the small side fits in memory, a broadcast is cheaper than any salt. Salt when the small side is too big to broadcast and AQE can't split the join, for example because the join is an outer join on the skewed side.

Match the random range to the exploded range. rand() returns values from 0 up to but not including 1, so floor(rand() * 64) gives 0 to 63, and the small side must carry exactly 0 to 63. A ceil on the big side produces 64 as a bucket that has no match, and an inner join silently drops those rows with no error.

Keep the types equal. A salt that is a long on one side and an int on the other can force a cast that stops the optimizer from using the key as a clean join condition. Cast both sides to the same type and drop the salt column after the join.

Do not salt a key that is skewed on both sides. The join output for a key is the product of its row counts on the 2 sides, and salting one side leaves the other side's hot rows to be copied N times. Salt the larger side and look for a way to pre-aggregate the other.

Check the result, not only the runtime. Compare the row count and a checksum of the salted join with the unsalted one on a sample, since every salting mistake that matters changes the output.

Isolation treats the hot keys as a different problem from everything else. Filter the big table into hot rows and the rest. Join the rest the normal way, where no partition is oversized now. Join the hot rows to just the matching rows of the small table with a broadcast, which is cheap because the hot keys are few. Union the 2 results. In the example the hot side is 40,000,000 rows against 1 customer row.

This is the fix for the case where the small side is too big to broadcast as a whole but its hot slice is tiny, and it avoids salting's copies and its random numbers. It needs the same list of hot keys, and it costs 2 scans of the big table unless you cache it.

Null keys are the easiest hot key to isolate. A null never equals anything in an equi-join, so null rows match nothing, but they still hash to the same partition. An inner join discards them, so filter them out before the shuffle. An outer join keeps them, so pull them out, give them their unmatched columns directly, and union them back.

A hot key hurts a group by less than a join, because Spark aggregates within each task before the shuffle. A count of 40 million rows for 1 key crosses the network as a few partial counts, one per task, and not 40 million rows. Skew bites a group by when the aggregate cannot be combined that way, such as a distinct count, a collected list or a percentile, and in window functions, where all rows of a partition key must be sorted together in 1 task.

The fix for an aggregate that can be combined in 2 steps is a salted 2 stage aggregation: group by the key and a salt, then group again by the key alone. It works for a sum of counts and doesn't work for a sum of distinct counts, since the same value can appear under 2 salts. For a window with a hot partition key, add a salt to the partition key only if the calculation can be split, and otherwise reduce the data first. See PySpark groupBy for how the shuffle behaves.

Skew also shows up when writing. The REBALANCE hint, documented in the Spark SQL hints reference, splits skewed partitions on a best effort basis before an output is written and is ignored when AQE is off. There is no SKEW hint in that reference, so a skew hint seen in a blog post belongs to some other system.

The question is usually phrased as a symptom: a job stuck on its last task, a join that runs for hours on a table with a popular key, or an out of memory error on 1 executor. A strong answer names the cause in terms of partitions and tasks, says how you would confirm it in the Spark UI, then walks the fixes in order of effort and says what each one costs.

Expect follow-ups on what you would do if the broadcast doesn't fit, how you would pick the salt count, and why salting breaks a distinct count. The PySpark interview questions and Spark optimization interview questions include this ground, the PySpark practice problems let you write the join, and a Spark mock interview lets you practice explaining it out loud.