How do I handle skewed joins in Spark?

Summary

• First check whether the join needs a shuffle: a broadcast hash join copies a small side to every worker, so the large side isn't shuffled by key and no hot key can pile into one task.
• For shuffle joins, Adaptive Query Execution (AQE) splits skewed partitions at runtime. It's on by default and treats a partition as skewed only if it's bigger than the median partition multiplied by 5.0 and also bigger than 256MB.
• AQE can only split certain sides: either side of an inner or cross join, the preserved side of a left or right outer join, the left side of semi and anti joins, and nothing in a full outer join.
• NULL keys never satisfy an equality join, yet on the preserved side of an outer join they still go through the shuffle, and every NULL hashes to the same value. Split those rows off and union them back.
• When AQE can't reach the skew, isolate the hot keys and broadcast-join only those rows; salt as a last resort.

How do I handle skewed joins in Spark?

Handle a skewed join in this order: avoid the shuffle when one side is small, let Adaptive Query Execution (AQE) split skewed partitions, confirm that your join type allows AQE to split the skewed side, and only then rewrite the query to isolate NULL or hot keys. Skew hurts because a shuffle join sends every row with the same key to the same task, so one hot key can keep a single task running long after the rest of the stage finishes. Most fixes are configuration and query shape.

Find the skew first

In the Spark UI's stage detail page, compare tasks on Shuffle Read Size / Records: a few tasks far above the median is the signature. Count rows per join key to find the hot keys and the NULLs. When AQE handles a join, the final plan shows it as SortMergeJoin(skew=true) or ShuffledHashJoin(skew=true).

Option	When to use it	Trade-off
Broadcast the small side	One side fits in memory on every worker	That side is copied to all workers
AQE skew join	Shuffle joins with a few oversized partitions	Splits only the sides the join type allows
Split off NULL keys	Outer joins where many keys are NULL	One more union in the plan
Isolate hot keys	A few known hot keys, including on a side AQE can't split	A second join and a hot-key list to maintain
Salting	Nothing else applies	Replicates the other side once per salt value

Step 1: broadcast, or let AQE split

spark.sql.autoBroadcastJoinThreshold (default 10485760 bytes, 10 MB) sets the largest table Spark broadcasts automatically, and the BROADCAST hint broadcasts the hinted side regardless of it. AQE can switch to a broadcast join at runtime, but the Spark docs say that isn't as efficient as planning it up front.
AQE skew handling needs both spark.sql.adaptive.enabled and spark.sql.adaptive.skewJoin.enabled, and both default to true. It flags a partition only if the partition is bigger than the median size multiplied by spark.sql.adaptive.skewJoin.skewedPartitionFactor (default 5.0) and bigger than spark.sql.adaptive.skewJoin.skewedPartitionThresholdInBytes (default 256MB). Keep that threshold above spark.sql.adaptive.advisoryPartitionSizeInBytes, as the docs advise. AQE drops a split that would add a shuffle unless spark.sql.adaptive.forceOptimizeSkewedJoin is true.

from pyspark.sql import functions as F

joined = orders.join(F.broadcast(countries), "country_code")

spark.conf.set("spark.sql.adaptive.skewJoin.skewedPartitionFactor", "3")
spark.conf.set("spark.sql.adaptive.skewJoin.skewedPartitionThresholdInBytes", "128MB")

Step 2: check what your join type allows

AQE's skew rule covers sort-merge and shuffled hash joins, and only certain sides:

Join type	Left side can be split	Right side can be split
Inner, cross	Yes	Yes
Left outer, left semi, left anti	Yes	No
Right outer	No	Yes
Full outer	No	No

In a left or right outer join, only the preserved side can be split. Hot keys on the null-supplying side need a broadcast of that side or hot-key isolation (Step 4).

Step 3: split off NULL keys

Under Spark's NULL semantics, an equality condition is never satisfied when a key is NULL. For inner and left semi joins, the optimizer infers filters on both sides, but on the preserved side of an outer join the NULL-key rows must survive, and they all hash to one partition. Join only the non-NULL rows, then add the rest back:

keyed = events.filter(F.col("user_id").isNotNull())
null_keyed = events.filter(F.col("user_id").isNull())

result = keyed.join(users, "user_id", "left").unionByName(
    null_keyed, allowMissingColumns=True
)

Step 4: isolate hot keys, then salt as a last resort

Join the dominant keys against a broadcast of their matching rows, then union with a normal join of the rest:

hot = [r.customer_id for r in orders.groupBy("customer_id").count()
       .orderBy(F.desc("count")).limit(20).collect()]
is_hot = F.col("customer_id").isin(hot)

hot_part = orders.filter(is_hot).join(F.broadcast(customers.filter(is_hot)), "customer_id")
rest = orders.filter(~is_hot).join(customers, "customer_id")
result = hot_part.unionByName(rest)

Salting appends a random suffix to the skewed side's key and replicates the other side once per suffix. It multiplies the replicated side, and if that side is preserved in an outer join, its unmatched rows come out once per salt, so use it last.

How Databricks approaches skewed joins

On Databricks, AQE is enabled by default and handles skew in sort merge and shuffle hash joins, with the same join-type limits. Databricks recommends relying on AQE skew join handling over the skew join hint, and the Spark UI marks a handled join by setting isSkew to true on its SortMergeJoin node.

FAQs

Why didn't AQE split my skewed partition?

Both size conditions must hold, the join must be a shuffle join on a splittable side, and the split mustn't add a shuffle unless you force it.

Does AQE handle skew in a full outer join?

No. Neither side can be split, and a broadcast hash join can't build either side of one. Rewrite it as a left outer join unioned with a left anti join of the other input; AQE can split the left side of both.

Should I still use a broadcast hint with AQE enabled?

Yes, when you know a side is small. AQE decides from runtime shuffle statistics, so it may switch only after both sides are shuffled; a planned broadcast skips that work.