How Spark's Adaptive Query Execution Handles Skewed Joins

If you have spent any real time running Spark in production, you know the pattern. A job that usually finishes in eight minutes is suddenly stuck at "199/200 tasks completed" for forty-five minutes. The cluster looks idle. One executor is pinned at 100% CPU, slowly chewing through something, while the other forty sit there doing nothing. Eventually it either finishes far too late or dies with an out-of-memory error.

That last straggler task is almost always the symptom of data skew in a join. And since Spark 3.0, there is a feature built specifically to deal with it without you rewriting a single line of business logic. It is controlled by spark.sql.adaptive.skewJoin.enabled, part of the broader Adaptive Query Execution (AQE) framework.

This article walks through what skew actually is, why it hurts so much, and the rather clever way AQE detects and repairs it at runtime. I will keep the examples concrete and the numbers small enough to follow on paper.

First, a quick refresher on how Spark joins data

To understand skew, you need to understand the sort-merge join, which is Spark's default strategy for joining two large tables.

Say you want to join an events table with a users table on user_id. Neither table is small enough to broadcast. Spark does the following:

1. Shuffle both sides by the join key. Every row is hashed on user_id, and rows that hash to the same value land in the same partition. The key guarantee here is that all rows for a given user_id end up together in one partition, on both sides.

2. Sort each partition by the key.

3. Merge the two sorted streams, emitting matched rows.

Here is the rough flow:

The important takeaway: a join partition is processed by one task. If that partition is enormous, one task has to do an enormous amount of work, and there is nothing the other tasks can do to help. They have already finished their small partitions and gone home.

So what is data skew?

Data skew is simply an uneven distribution of rows across partitions. In a perfect world every partition after a shuffle is about the same size. In the real world, some keys are far more common than others.

The textbook example is a "catch-all" key. Imagine our events table is web analytics data, and anonymous visitors are all recorded as user_id = 0. Logged-in users each generate a few hundred events. The anonymous bucket generates two hundred million.

When Spark shuffles on user_id, this is what the partition sizes look like:

Partition 0   ██ 100 MB
Partition 1   ██ 110 MB
Partition 2   ██  95 MB
Partition 3   ████████████████████████████████████  2000 MB   ← user_id = 0 lives here
Partition 4   ██ 105 MB
Partition 5   ██  98 MB
...

Partition 3 is twenty times larger than its neighbours. The task that processes it will take roughly twenty times longer, hold twenty times more data in memory, and very possibly spill to disk or fall over entirely. Meanwhile your other 199 cores finished ages ago and are burning money doing nothing.

This is the straggler problem, and it is the single most common reason a Spark stage runs slowly for no obvious reason.

Why you can't just "add more partitions"

A reasonable first instinct is to bump up spark.sql.shuffle.partitions. It does not help. All two hundred million user_id = 0 rows hash to the same value, so they go to the same partition no matter how many partitions you create. You would just end up with more empty partitions and the same one giant one.

The traditional fix was salting - appending a random suffix to the hot key so it spreads across several partitions, then doing extra work to merge the results back. It works, but it is fiddly, it pollutes your query, and you have to know in advance which keys are skewed. That last part is the killer: skew often shifts day to day.

This is exactly the problem AQE was built to solve.

Enter Adaptive Query Execution

The classic Spark optimizer (Catalyst) plans the entire query before execution, based on whatever statistics it has. The trouble is that those statistics are often missing, stale, or simply impossible to know ahead of time and the size of a shuffle partition is a perfect example of something you only discover once the data is actually moving.

AQE changes this. It lets Spark re-optimise the plan at runtime, using real statistics gathered during execution. After each shuffle's map stage completes, Spark has precise size information for every single partition. AQE uses that fresh data to make three kinds of adjustment:

• Coalescing lots of tiny partitions into fewer, sensibly sized ones.
• Switching join strategy - for example, flipping a sort-merge join to a broadcast join once it learns a table is actually small.
• Splitting skewed partitions in a join, the subject of this article. The whole point is that AQE reacts to what really happened, not to what the optimiser guessed would happen.

The whole point is that AQE reacts to what really happened, not to what the optimiser guessed would happen.

How AQE detects a skewed partition

This is where it gets nicely pragmatic. After the map stage of a join finishes, AQE looks at the list of partition sizes and flags a partition as skewed when both of these conditions are true:

1. The partition is larger than spark.sql.adaptive.skewJoin.skewedPartitionThresholdInBytes (default 256 MB), and
2. The partition is larger than spark.sql.adaptive.skewJoin.skewedPartitionFactor (default 5.0) times the median partition size. Both conditions matter, and the reasoning behind each is worth understanding:

• The factor catches partitions that are abnormal relative to their peers. If every partition is roughly 1 GB, then no single one is skewed. they are all just big, and splitting one wouldn't rebalance anything. The factor only triggers when something genuinely stands out.

• The threshold prevents pointless splitting when everything is small. Suppose the median partition is 5 MB and one partition is 40 MB. That is 8× the median, which clears the factor, but 40 MB is trivial to process. Splitting it would add overhead for no benefit. The absolute threshold says "don't bother unless it is actually large." Going back to our example: the median is about 100 MB, and the skewed partition is 2000 MB.

• Is 2000 MB > 256 MB? Yes.

• Is 2000 MB > 5 × 100 MB = 500 MB? Yes. Both hold, so Partition 3 is marked as skewed.

How AQE fixes it: split and replicate

Here is the elegant part. Once AQE knows Partition 3 on the events side is skewed, it splits that one partition into several smaller slices, each roughly the size of spark.sql.adaptive.advisoryPartitionSizeInBytes (default 64 MB; I'll use 256 MB in this example to keep the arithmetic clean).

A 2000 MB partition split into 256 MB slices gives us about eight slices. Each slice can now be processed by its own task, in parallel.

But there is a catch with joins. Splitting the big side alone is not enough, because every row on the events side still needs the chance to match against the corresponding rows on the users side. So AQE replicates the matching partition on the other side of the join, handing a full copy to each slice.

![image]

You might worry that replicating the other side eight times is expensive. In practice it almost never is, and here is why: skew is asymmetric. The events side has two hundred million rows for user_id = 0; the users side has exactly one row for it (plus whatever other users happened to hash into the same partition a few megabytes at most). Copying a 50 MB partition eight times costs you 400 MB of extra shuffle reads, which is nothing next to the 2 GB monster you just defused.

So the trade is overwhelmingly in your favour: one task doing 2 GB of work becomes eight tasks doing 256 MB each, finishing in roughly an eighth of the time, with a tiny, well-spent overhead.

Before and after, side by side

	Without skew join handling	With skew join handling
Tasks for the hot partition	1	~8
Work per task	2000 MB	~256 MB
Wall-clock for the stage	Dominated by one slow task	Limited by the slowest of eight balanced tasks
Memory pressure on one executor	Very high, likely spills/OOM	Spread out, comfortable
Idle cores while the straggler runs	Most of the cluster	None
Code changes required	None (you'd reach for salting)	None

---

The configuration knobs

Everything here is driven by configuration. The defaults are sensible for most workloads, but it helps to know what each one does.

Configuration	Default	What it controls
spark.sql.adaptive.enabled	true (since Spark 3.2.0)	The master switch for AQE. Nothing below works without it.
spark.sql.adaptive.skewJoin.enabled	true	Turns on skew join handling specifically.
spark.sql.adaptive.skewJoin.skewedPartitionFactor	5.0	How many times the median a partition must be to count as skewed.
spark.sql.adaptive.skewJoin.skewedPartitionThresholdInBytes	256MB	The absolute size a partition must also exceed before it is split.
spark.sql.adaptive.advisoryPartitionSizeInBytes	64MB	The target size for each slice when a skewed partition is broken up.

A quick note on versions: in Spark 3.0 and 3.1, AQE shipped disabled by default, so you had to set spark.sql.adaptive.enabled=true yourself. From 3.2.0 onward it is on by default, and so is skew join handling. If you are on an older runtime, double-check that first switch, it is the most common reason people think the feature "isn't working."

Turning it on explicitly is as simple as:

spark.conf.set("spark.sql.adaptive.enabled", "true")
spark.conf.set("spark.sql.adaptive.skewJoin.enabled", "true")

If your skew is severe, lowering skewedPartitionFactor (say to 3) or reducing advisoryPartitionSizeInBytes makes Spark split more aggressively into smaller pieces. Tune by observation, not by guesswork.

The benefits of enabling it

Pulling the threads together, here is what you actually gain:

It kills stragglers. This is the headline benefit. Stage time stops being held hostage by a single oversized task, and your wall-clock time often drops dramatically on skewed workloads.

It prevents out-of-memory failures. A 2 GB partition that one executor has to sort and hold in memory is a prime candidate for spilling to disk or crashing outright. Splitting it keeps each task's footprint small and predictable.

It uses your cluster properly. Instead of one core working while the rest idle, the work is spread across many cores. You are paying for the whole cluster this lets you actually use it.

It needs zero code changes. This is the part data engineers appreciate most. No salting, no random-suffix gymnastics, no merge-back logic cluttering your transformations. You flip a config and the engine handles it.

It adapts to reality at runtime. You do not need to know which keys are skewed in advance, and it keeps working even when the skewed key changes from one day to the next which, in practice, it often does. Salting can't do that without you re-profiling the data.

It replaces a whole category of manual hacks. Much of the skew-handling folklore from the Spark 2.x era salting, custom partitioners, splitting jobs in two is simply unnecessary now for join skew. That is a meaningful reduction in maintenance burden.

---

Limitations and honest caveats

No feature is magic, and it is better to know the edges up front.

It only helps shuffle-based joins. Skew join handling applies to sort-merge joins (and shuffle-hash joins). A broadcast join has no shuffle and no per-partition tasks in the same sense, so this optimisation doesn't apply to it though broadcast avoids the skew problem in its own way by sending the small table everywhere.

It can't split below map-output granularity. When Spark splits a skewed partition, it does so by grouping the contributions of individual map tasks. If a single map task already produced a huge chunk for the hot key, AQE cannot slice finer than that chunk. In extreme cases this limits how balanced the result can be.

Some join types have restrictions. Inner joins are fully supported. For outer joins, only the safe side can be split (you cannot freely split the side whose rows must be preserved), so the optimisation may apply to one side only or not at all depending on the join type.

It does not fix aggregation skew. This feature is specifically about joins. If your skew is in a GROUP BY say, counting events per user_id where one user dominates that is a different problem. AQE's partition coalescing helps to a degree, but heavy aggregation skew may still call for salting or a two-step aggregation.

It is reactive, not free. Detection happens after the map stage completes, because that is when the real statistics exist. There is a small planning cost, and the replication adds some shuffle reads. For the workloads that need it, this is a rounding error but it is not literally zero.

---

How to confirm it actually kicked in

You don't have to take it on faith. A few ways to verify:

The Spark UI, SQL/DataFrame tab. Open the query's plan in the UI. The shuffle-read node (shown as AQEShuffleRead) reports how many partitions were treated as skewed and split. If you see a non-zero skewed-partition count there, the feature did its job.

The physical plan. An adaptive query shows up as AdaptiveSparkPlan. Before execution it is marked isFinalPlan=false; once it has run and re-optimised, the final plan is marked isFinalPlan=true, and you can read off the actual partition handling.

The task count for the stage. A clear tell is seeing more tasks than your configured shuffle partitions for the join stage. Those extra tasks are the slices AQE created from the skewed partitions.

Task duration distribution. In the stage's event timeline, a healthy run shows tasks of broadly similar length. If skew handling is off, you'll see that one long bar stretching far past all the others the straggler. With it on, the bars even out.

---

Wrapping up

Data skew in joins is one of those problems that feels mysterious the first time you hit it the job isn't broken, it's just inexplicably slow, and only sometimes. Underneath, the cause is almost always boringly simple: one key is far more popular than the rest, all its rows pile into a single partition, and one unlucky task carries the whole load.

Spark's adaptive skew join handling tackles this exactly where it should at runtime, using real statistics, by splitting the offending partition into manageable pieces and replicating the small side to keep the join correct. The result is balanced parallelism, lower memory pressure, and faster jobs, all from a configuration flag rather than a rewrite.

If you are on Spark 3.2 or later, it is already on. The most valuable thing you can do is learn to recognise skew in the Spark UI, confirm the feature is engaging, and reserve manual tricks like salting for the cases it genuinely can't reach aggregation skew, mostly. For everything else, let the engine do the work.