How do I fix data skew (salting, AQE skew join)?

Summary

• Data skew means a few join or grouping keys hold a disproportionate share of the rows, so a few tasks in the shuffle stage run much longer than the others. Confirm it in the Spark UI (per-task duration and shuffle read size) or by counting rows per key.
• Start with Adaptive Query Execution (AQE) skew join. AQE has been enabled by default since Spark 3.2.0, and spark.sql.adaptive.skewJoin.enabled defaults to true. It splits skewed partitions of shuffled joins and replicates the other side if needed.
• A partition counts as skewed only if it is larger than spark.sql.adaptive.skewJoin.skewedPartitionFactor (default 5.0) times the median partition size and also larger than spark.sql.adaptive.skewJoin.skewedPartitionThresholdInBytes (default 256MB).
• Salting is the manual fallback: add a random integer to the key on the skewed side, copy the other side once per salt value, and join on key plus salt. The Databricks optimization guide ranks it last because it requires code changes.
• The AQE skew-join setting is documented for shuffled joins. For a skewed aggregation, the same guide salts only the values that cause the skew.

How do I fix data skew (salting, AQE skew join)?

Data skew means a few keys hold a disproportionate share of the rows, so a few tasks in a shuffle stage run long while the others finish and wait. Try the fixes in this order: drop junk keys, let Adaptive Query Execution (AQE) skew join split the hot partitions, and salt the key yourself only if that isn't enough. The Spark tuning guide puts the stakes plainly: "Data skew can severely downgrade the performance of join queries."

How do I confirm it's skew?

Open the slow stage in the Spark UI. The stage detail page shows summary metrics for task duration and shuffle read size, plus a per-task table. The Databricks Spark UI guide says to compare the maximum task duration with the 75th percentile duration, and a maximum far above it points to skew. Then count rows per key. The Databricks optimization guide calls "an enormous disparity between the row counts" a definite skew:
SELECT join_key, COUNT(*) FROM t GROUP BY join_key ORDER BY COUNT(*) DESC LIMIT 20

How does AQE skew join decide what to split?

AQE re-optimizes the plan at runtime and has been "enabled by default since Apache Spark 3.2.0". With the skew rule on, Spark "dynamically handles skew in shuffled join (sort-merge and shuffled hash) by splitting (and replicating if needed) skewed partitions".

Setting	Default	Meaning
spark.sql.adaptive.enabled	true	Umbrella switch for AQE
spark.sql.adaptive.skewJoin.enabled	true	Turns the skew rule on
spark.sql.adaptive.skewJoin.skewedPartitionFactor	5.0	Skewed if larger than this factor times the median partition size
spark.sql.adaptive.skewJoin.skewedPartitionThresholdInBytes	256MB	And also larger than this many bytes
spark.sql.adaptive.forceOptimizeSkewedJoin	false	Apply the rule even if it adds a shuffle

Both size tests must pass. The docs say the byte threshold should ideally be larger than spark.sql.adaptive.advisoryPartitionSizeInBytes (64 MB by default).

# Flag partitions as skewed sooner (defaults: 5.0 and 256MB)
spark.conf.set("spark.sql.adaptive.skewJoin.skewedPartitionFactor", "3.0")
spark.conf.set("spark.sql.adaptive.skewJoin.skewedPartitionThresholdInBytes", "128MB")

# Apply the skew rule even if it adds a shuffle (default false)
spark.conf.set("spark.sql.adaptive.forceOptimizeSkewedJoin", "true")

How does salting work?

The Databricks optimization guide defines salting as "breaking a large skewed partition into smaller partitions by appending random integers as suffixes to skewed column values". In a join, the skewed side gets a random salt and the other side is copied once per salt value:

from pyspark.sql import functions as F

SALTS = 8  # buckets to spread each hot key over

orders_s = orders.withColumn("salt", (F.rand() * SALTS).cast("int"))

salts = spark.range(SALTS).select(F.col("id").cast("int").alias("salt"))
customers_s = customers.crossJoin(salts)

result = orders_s.join(customers_s, ["customer_id", "salt"], "left").drop("salt")

crossJoin returns the cartesian product, so the number of buckets trades spreading the hot key against duplicating the other side. For a skewed aggregation, aggregate by key plus salt first, then aggregate the partial results by key. The guide adds that you only need to salt the values that contribute to the skew.

Which fix should I try first?

Option	Use it when	Trade-off
Filter the key	The skew comes from rows you don't need, such as nulls	Changes the result if those rows matter
AQE skew join	Shuffled joins, as the default first choice	No code change, but only partitions past both thresholds are split
BROADCAST hint	One side is small enough to send to every worker	Spark doesn't guarantee it will honor the hint
Salting	AQE isn't enough, or the skew is in an aggregation	Code changes and a copy of the other side per salt value

The tuning guide says AQE's runtime switch to a broadcast join "is not as efficient as planning a broadcast hash join in the first place".

How does Databricks handle data skew?

AQE is on by default on Databricks. Its docs list this AQE feature: "Dynamically handles skew in sort merge join and shuffle hash join by splitting (and replicating if needed) skewed tasks into roughly evenly sized tasks." They show spark.sql.adaptive.skewJoin.enabled, skewedPartitionFactor and skewedPartitionThresholdInBytes with the same defaults. The retired skew-hint docs say "Skew join hints are not required. Databricks handles skew by default by using adaptive query execution (AQE)." To confirm a split, open the finished query in the Spark UI: the AdaptiveSparkPlan node's isFinalPlan flag is true, and a handled skew join shows as a SortMergeJoin node with isSkew true.

FAQs

Does AQE skew join fix a skewed GROUP BY?

The skew-join setting is documented for shuffled joins, not aggregations. For a skewed aggregation, the Databricks optimization guide uses partial salting.

What if the skewed key is null?

The same guide says that if you join on a column with many nulls, "filtering out the null values will resolve the issue". Do that only if the null rows don't contribute to the result.

Why isn't AQE splitting my skewed partition?

A partition must be larger than the factor times the median partition size and also larger than the byte threshold, so with the default threshold a partition under 256MB is never split. spark.sql.adaptive.enabled and spark.sql.adaptive.skewJoin.enabled must both be on, and the join must be a shuffled join.