Spark Shuffle

Almost everything written about the Spark shuffle blames the network. Network bandwidth is rarely what you are waiting for. You are waiting for one task — and a stage ends when its slowest task ends, not when the average one does.

Below is a stage with 12 map tasks writing into a shuffle, and 8 executor cores reading it back. Nothing about the query changes as you move these controls. Only the distribution of the join key changes.

Push skew to 100 and watch the two numbers separate. Total CPU goes from 33.1s to 60.8s — under twice the work, and most of that increase is one task spilling to disk. Stage wall clock goes from 4.7s to 46.1s, nearly tenfold, because the work stopped being divisible. At skew 0 the eight cores share 33.1s of work and finish in 4.7s. At skew 100, 46.1s of the 60.8s is a single task; the remaining 14.7s spreads over the other seven cores, which are done in about two seconds each and then idle for three quarters of a minute. Look at the timeline: seven empty lanes.

Why more partitions stops helping

The instinctive fix is to raise spark.sql.shuffle.partitions (default 200; the simulation uses a smaller range so you can see individual tasks). Set skew to 0 and walk the partition slider up: 15.4s at 4, 6.8s at 8, 4.7s at 16 — then it reverses, 5.0s at 32 and 6.5s at 64. Past the point where every core has work, each extra partition is another task to launch, track and fetch.

Now set skew to 100 and walk the same slider. 46.5s at 4, 46.1s at 16, 46.0s at 64. Sixteen times the partitions, half a second of difference.

Partitioning is hash(key) % numPartitions. Adding partitions divides the keys into more buckets, but a single key is atomic: it hashes to exactly one partition no matter how many there are. If one key holds 40% of the rows, then one partition holds 40% of the rows at 16 partitions and at 16,000. This is the single most important fact about shuffle skew, and it is why repartition(2000) is not a fix — see how Spark decides which row goes where.

Meanwhile the cost of more partitions is real, and it is not mainly network. Every map task writes one block per reduce partition, so the number of blocks is the product of the two. The simulation below spreads a 400 GB shuffle over that product:

At 200 × 200 the average block is 10.0 MB and a reducer fetching it gets a sequential read. At 2,000 × 2,000 it is four million blocks averaging 100 KB — still not tiny, but each one is a separate entry in the map output tracker and a separate seek on the reader's side, and four million of them is four million round trips through the shuffle service. Push either slider further in a real job and the metadata starts to outweigh the data. This is the other end of the same trade-off: too few partitions and each task spills, too many and you drown in bookkeeping.

Spill is not a failure mode, it is a budget

Put the partition slider back to 16 and walk skew up from 0. The spill readout says none through skew 20 and lights amber at skew 30, at 1 of 16. A reduce task builds its output in memory; when it exceeds what the executor gives it, it sorts what it has, writes it to local disk, and carries on. The task still succeeds — it just got several times slower, and the slowdown is invisible in anything except task-level metrics.

Watch wall clock as you keep going. Skew 30 costs 4.9s with the largest task reading 31M rows. Skew 40 costs 9.3s with the largest reading 47M. Rows in that task rose by half; wall clock nearly doubled. The extra factor is the spill: at 47M rows the task is 21M rows past what it can hold, and those rows get sorted, written to local disk and read back. The curve is not smooth, and the discontinuity is not in the data — it is in a threshold.

This is why a skewed stage often shows up first as a mysterious 10× runtime regression rather than an error. Nothing failed. One task crossed a memory threshold and started paying disk prices, and because the stage clock is a maximum, that one task became the whole stage. Only when spill exhausts disk, or the task's memory request exceeds the executor's total, do you get the familiar ExecutorLostFailure and Container killed by YARN for exceeding memory limits — the errors people then try to fix by raising executor memory, which buys headroom without touching the cause.

Salting: make the atom smaller

Since the problem is that a key is indivisible, the fix is to make the key divisible. Put the partition slider back to 16, set skew to 100, and tick Salt the hot key. Wall clock drops from 46.1s to 10.7s.

Salting appends a random suffix to the hot key on one side of the join and explodes the other side to match, turning user_id = 0 into 0#0 … 0#3. Four keys instead of one, so four hash values, so up to four partitions and four tasks. Nothing about the mechanism changed — the partitioner still sends every row of a key to exactly one reducer. You changed what counts as a key.

The usual warning about salting is that it costs CPU, and this simulation disagrees with it. Total CPU goes down, from 60.8s to 47.4s. The reason is the spill readout: the 174M-row task was 148M rows over the in-memory budget and paying disk prices for all of them, while four 44M-row tasks each spill far less. When your straggler is over the spill threshold, salting is a throughput win as well as a latency win.

The cost the warning is really about is one this model does not track: it follows the skewed side of the join only. In a real salted join the other side must be exploded by the salt factor for the salted keys, so you pay an extra n copies of the matching rows. If the other side is a small dimension table that is nothing; if it is a second fact table it is the dominant term. Measure it before assuming either.

What AQE fixes, and where it gives up

Leave skew at 100 and partitions at 16, untick salt, and tick Enable AQE skew join. Wall clock goes from 46.1s to 5.1s and the spill readout goes to none. Adaptive query execution — AQE, Spark's re-planning of a query against the statistics it measures while running it — did automatically what you just did by hand, on a partition holding one key.

That last part is the thing most write-ups get wrong, including an earlier version of this lesson. The usual explanation is that AQE splits a partition by slicing the map-side blocks that feed it, so a partition containing a single key has nothing to slice, because every row of a key must reach the same reducer. The first half is right and the conclusion does not follow. Spark's rule is OptimizeSkewedJoin, and its own description of what it does is: "divide each skew partition into smaller partitions and replicate its matching partition on the other side of the join so that they can run in parallel tasks."

The replication is the whole trick. Sub-task 3 of 12 gets a slice of the hot key's rows and a full copy of the other side's matching partition, so it can join its slice correctly on its own. Which is, to a first approximation, salting — done by the planner, from measured statistics, without you touching the query. That is why the decision log under the chart says the split succeeded: this model does the same thing.

So where does it actually give up? Move the Shuffle control and watch the log.

• groupBy aggregation. Wall clock goes straight back to 46.1s. OptimizeSkewedJoin is a join rule: it lives in the adaptive optimiser's join handling and there is no "other side" to replicate. A single hot key in a groupBy, reduceByKey or window shuffle is genuinely indivisible, and salting (with a two-stage partial aggregation) is the only fix.
• Full outer join. Also 46.1s. The rule can split the left side of an inner, cross, left semi, left anti or left outer join, and the right side of an inner, cross or right outer join. A full outer join appears in neither list, because replicating either side would duplicate the unmatched rows that side is there to contribute.
• Below the thresholds. Set skew to 0 with AQE on and the log reads "not skewed by Spark's definition". A partition qualifies only if it is both larger than skewedPartitionFactor × the median partition and larger than skewedPartitionThresholdInBytes — 5.0 and 256 MB by default in Spark 4.x. A 900 MB partition next to a 400 MB median fails the factor test; a 40 MB partition next to a 1 MB median fails the byte test. This is why teams turn skew join on, see no change, and conclude it is broken.
• Both sides skewed on the same key. Not in this simulation, and the reason is structural: the rule splits one side and replicates the other, and if the other side is also enormous for that key you have multiplied it by the split factor.

One more ceiling. Set Shuffle back to inner join and skew back to 100: the split in the log reads "12 map-block ranges", and 12 is the number of map tasks in this stage. You cannot cut a partition into more pieces than there are map-side blocks feeding it. A hot key arriving from a small number of large map tasks has a hard floor on how far it can be divided.

AQE and skew join are both on by default — spark.sql.adaptive.enabled and spark.sql.adaptive.skewJoin.enabled default to true (Spark 3.2 and later; verified against SQLConf.scala on master, 2026-08-21). If your skewed join is still slow, the question is not "should I enable it" but which of the four rows above you are standing in. The lesson on AQE walks through the thresholds directly.

Reading a real stage

In the Spark UI, open the stage and sort tasks by duration. Look at the summary metrics table, specifically the max versus 75th percentile row for Shuffle Read Size / Records. A max that is ten times the 75th percentile is skew, full stop — you do not need to guess. Then check Spill (disk) on that same row: if the long task spilled and the others did not, you have located both the cause and the multiplier.

Related mechanisms worth having in your head before you tune anything: broadcast joins avoid the shuffle entirely when one side is small enough, and consumer group rebalancing is the same "one slow member stalls the group" shape in a completely different system.