What is data skew?

Summary

• Data skew is uneven data distribution across partitions. In a shuffle, all rows for a key are processed together, so a few frequent keys create a few very large tasks that take much longer than the rest and hold up the whole stage.
• Adaptive Query Execution (AQE) has been on by default since Apache Spark 3.2.0. Its skew join feature splits skewed tasks in sort-merge joins into roughly evenly sized tasks, replicating if needed.
• A partition counts as skewed only if it is larger than both the skewed-partition factor times the median partition size (default 5.0) and the byte threshold (default 256MB).
• To detect skew, open a slow stage in the Spark UI and compare the maximum task duration with the 75th percentile, then count rows per key in the join or group-by column.
• When AQE doesn't help, broadcast the small join side, pre-aggregate, or salt the key. Repartitioning by the hot key does not split it.

What is data skew?

Data skew is an uneven distribution of data across partitions. Spark's shuffle regroups rows so that all values for a key are processed together. If a few key values account for most of the rows, the tasks that receive them run much longer than the rest. Databricks' Spark UI guide describes skew as one or just a few tasks taking much longer than the rest, which results in poor cluster utilization and longer jobs. The Spark docs add that data skew can severely downgrade the performance of join queries.

What causes data skew?

Cause	What happens	Example
Hot keys	A few key values account for most of the rows, and each key's rows meet in one task	One customer_id owns most of the orders
NULL keys	GROUP BY and DISTINCT put all NULL values into one bucket	A mostly empty nullable column
Dominant categories	Rows cluster on a few values of a column	One country supplies most of the events

How do you detect data skew?

Open the slow stage in the Spark UI. The stage detail page shows summary metrics for all tasks, including task duration and shuffle read size and records. Databricks suggests comparing the maximum duration with the 75th percentile: a maximum much higher than the 75th percentile points to skew.
Then count rows per key in the join or group-by column:

SELECT customer_id, COUNT(*) AS row_count
FROM orders
GROUP BY customer_id
ORDER BY row_count DESC
LIMIT 10;

If a handful of keys dominate the counts, joins and aggregations on that column will skew.

How does Adaptive Query Execution fix skew?

AQE re-optimizes a query plan in the middle of execution, using runtime statistics. Its skew join feature "dynamically handles skew in sort-merge join by splitting (and replicating if needed) skewed tasks into roughly evenly sized tasks."

Setting	Default	Role
spark.sql.adaptive.enabled	true	Umbrella switch for AQE
spark.sql.adaptive.skewJoin.enabled	true	Turns on skew join handling
spark.sql.adaptive.skewJoin.skewedPartitionFactor	5.0	Skewed only if larger than this factor times the median partition size
spark.sql.adaptive.skewJoin.skewedPartitionThresholdInBytes	256MB	And also larger than this size
spark.sql.adaptive.advisoryPartitionSizeInBytes	64MB	Target size when skewed partitions are split

Both size conditions must hold. If AQE misses skew you can see in the Spark UI, lower the byte threshold. The docs advise keeping it above the advisory size:

SET spark.sql.adaptive.skewJoin.skewedPartitionThresholdInBytes=128MB;

What can you do when AQE isn't enough?

• Broadcast the small side. A BROADCAST hint makes Spark broadcast the hinted side regardless of spark.sql.autoBroadcastJoinThreshold. AQE may switch to a broadcast join only after it has shuffled both sides, so a hint you know is right can save that work. The left side of a left outer join can't be broadcast.
• Pre-aggregate. If the join needs only aggregated values, aggregate by the join key first. Databricks recommends "key salting or pre-aggregation" for uneven data.
• Salt the key. For a hot key AQE can't split, add a random salt to the big side and replicate the other side once per salt value:

from pyspark.sql import functions as F

n = 8  # number of salt buckets
big = big_df.withColumn("salt", (F.rand() * n).cast("int"))
small = small_df.withColumn(
    "salt", F.explode(F.sequence(F.lit(0), F.lit(n - 1)))
)
joined = big.join(small, ["customer_id", "salt"]).drop("salt")

• Rebalance output. The REBALANCE hint splits skewed output partitions on a best-effort basis, which helps when you write a result to a table. It targets output partitions and file sizes, not the join itself, and it is ignored if AQE is disabled.
• Avoid REPARTITION(customer_id). It hash-partitions by that column, so every row with the hot value still lands in one partition.

How Databricks approaches data skew

AQE is enabled by default on Databricks, with the same skew defaults of 5 and 256MB. Databricks documents that its skew handling applies only to shuffle-based joins (sort merge and shuffle hash joins), never to broadcast joins, and that the join type decides which side can be optimized: a full outer join has no skew-optimized side. A skew-handled join appears in the plan as a SortMergeJoin node with isSkew set to true. Databricks recommends relying on AQE rather than the older skew join hint, and Databricks SQL flags a DATA_SKEW insight on queries with uneven data distribution.

FAQs

Does AQE always fix skew?

No. The documented feature targets joins, both size conditions must be met, and on Databricks the join type limits which side is optimized. For skew in a group-by, use pre-aggregation or salting.

Why didn't AQE detect my skew?

Databricks lists two conditions: the partition must exceed the byte threshold and also exceed the median partition size times the skewed-partition factor. A skewed partition under either limit is left alone.

Does the salt make the join more expensive?

Yes. The replicated side grows n-fold, one copy per salt value, so keep n small and use salting only for keys that stay skewed after AQE and broadcasting.