Spark Data Skew: Diagnosing It Properly, Then Fixing It With Salting, AQE, and Broadcast

The Spark job runs for 41 minutes. The stage detail says 400 tasks: 399 finished in under 30 seconds, one is still going. Your cluster is 39 executors sitting idle while a single core grinds through 90 GB of rows that all share one key.

That is skew, and it is the most common performance problem I see after accidental shuffles. It is also the one most often misdiagnosed, because the symptom — a long stage — looks identical to “the cluster is too small.” Teams double the cluster, the stage takes exactly as long, and they conclude Spark is slow.

Skew is not a tuning problem. It is a data distribution problem that shows up at every shuffle boundary, and the fix depends entirely on why the distribution is lopsided. A NULL-heavy foreign key, a customer_id = -1 sentinel, and a genuine enterprise customer with 200 million events are three different problems with three different answers.

Here is how I identify which one I have, in about five minutes, and then the four fixes in the order I reach for them.

Diagnosis: three queries and one UI screen

Start with the UI, because it tells you whether skew is even the issue. Stages tab, click the slow stage, look at the task metrics summary table. Spark gives you min, 25th percentile, median, 75th, and max for duration, shuffle read, and spill.

The ratio I use is max duration divided by median duration. Under 3 is normal variance. Above 5 is skew worth fixing. Above 20 means one key is your entire runtime. Check shuffle read bytes on the same row — if the max task read 90 GB while the median read 180 MB, you are not guessing anymore.

Then find the key. This is a cheap query and I run it before any fix:

from pyspark.sql import functions as F

events = spark.read.table("prod.silver.events")

# Top keys by row count: the skew census
(events.groupBy("customer_id")
       .count()
       .orderBy(F.desc("count"))
       .limit(20)
       .show(truncate=False))

# What fraction of the table is the single worst key?
total = events.count()
worst = (events.groupBy("customer_id").count()
               .orderBy(F.desc("count")).first())
print(f"{worst['customer_id']}: {worst['count'] / total:.2%} of rows")

# NULL and sentinel check, which is usually the real answer
(events.select(
    F.count(F.when(F.col("customer_id").isNull(), 1)).alias("nulls"),
    F.count(F.when(F.col("customer_id") == -1, 1)).alias("neg_one"),
    F.count(F.when(F.col("customer_id") == "", 1)).alias("empty"),
).show())

In my experience, three quarters of skew incidents end at that last query. NULL foreign keys from a source system that allows unmatched records, or a -1 / UNKNOWN sentinel inserted by an upstream ETL step to avoid nulls, which is worse because it looks like a real key.

flowchart TD
  slow["Stage with one long task"] --> ratio{"Max / median duration > 5?"}
  ratio -->|no| notskew["Not skew: check shuffle volume"]
  ratio -->|yes| census["Top 20 keys by count"]
  census --> kind{"What is the hot key?"}
  kind -->|"NULL or sentinel"| filt["Filter or route before join"]
  kind -->|"one large real key"| salt["Salt the hot keys only"]
  kind -->|"small dimension side"| bcast["Broadcast join"]
  kind -->|"moderate, many keys"| aqe["AQE skew join"]

The fix depends on why the key is hot. Diagnose first, tune second.

Fix one: filter or route the NULL keys

If the hot key is NULL, the rows are not joining to anything anyway. An inner join drops them after paying full shuffle cost for them; a left join produces all-null right-side columns. Either way, you can decide their fate before the exchange.

-- Before: 340 million NULL customer_ids all hash to one partition
SELECT e.event_id, c.segment
FROM prod.silver.events e
LEFT JOIN prod.gold.dim_customer c USING (customer_id);

-- After: split the paths, only join what can match
WITH matchable AS (
  SELECT * FROM prod.silver.events WHERE customer_id IS NOT NULL
),
unmatchable AS (
  SELECT * FROM prod.silver.events WHERE customer_id IS NULL
)
SELECT m.event_id, c.segment
FROM matchable m
LEFT JOIN prod.gold.dim_customer c USING (customer_id)
UNION ALL
SELECT u.event_id, CAST(NULL AS STRING) AS segment
FROM unmatchable u;

Two shuffles become one small one plus a scan. On a pipeline I inherited last year this took a nightly stage from 38 minutes to 4. It also surfaced a data quality question nobody had asked: why were 12 percent of events unattributed? That belongs in a data quality check with an alert, not silently absorbed by a join.

The sentinel version is the same shape, except you cannot filter — customer_id = -1 may be semantically meaningful downstream. Route it to a constant-join path instead of the real join.

Fix two: let AQE do it

Adaptive Query Execution has a skew join optimization that splits oversized shuffle partitions into smaller ones and replicates the matching side. It is on by default in Spark 3.2 and later, and when it applies it is entirely free.

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

# A partition is "skewed" only if BOTH are true:
#   size > skewedPartitionThresholdInBytes
#   size > median partition size * skewedPartitionFactor
spark.conf.set("spark.sql.adaptive.skewJoin.skewedPartitionThresholdInBytes", "256MB")
spark.conf.set("spark.sql.adaptive.skewJoin.skewedPartitionFactor", "5")
spark.conf.set("spark.sql.adaptive.advisoryPartitionSizeInBytes", "128MB")

The two conditions are the whole story, and where teams get surprised. Defaults are 256 MB and a factor of 5. If your median partition is 400 MB, a 1.6 GB partition is not considered skewed because it is only 4x the median. Lowering the factor to 3 and the threshold to 128 MB catches considerably more real-world skew, which is what I run in production.

The hard limit: AQE splits a partition into sub-ranges of the shuffle output. It cannot split a single key, because all rows for that key must meet. If one customer_id is 90 GB, AQE gives you one 90 GB task no matter how you configure it. Confirm from the SQL tab — the plan node will read SortMergeJoin ... isSkewJoin=true with a skew partition count when it kicked in.

Fix three: salt the hot keys, and only the hot keys

When one key is genuinely huge and genuinely needed, you have to break it apart manually. Salting adds a random suffix to the join key on the large side and explodes the small side to match, turning one partition into N.

The naive version salts everything, which multiplies your dimension table by N and adds a shuffle for the tail. I only salt the keys I measured as hot:

from pyspark.sql import functions as F

HOT = ["cust_881204", "cust_119007"]   # from the census query above
SALT_N = 64

events = spark.read.table("prod.silver.events")
dim    = spark.read.table("prod.gold.dim_customer")

is_hot = F.col("customer_id").isin(HOT)

# Hot path: spread each hot key across SALT_N partitions
hot_events = (
    events.where(is_hot)
          .withColumn("salt", (F.rand() * SALT_N).cast("int"))
          .withColumn("join_key", F.concat_ws("#", "customer_id", "salt"))
)

# Explode only the hot dimension rows, once per salt value
hot_dim = (
    dim.where(is_hot)
       .withColumn("salt", F.explode(F.array([F.lit(i) for i in range(SALT_N)])))
       .withColumn("join_key", F.concat_ws("#", "customer_id", "salt"))
       .drop("customer_id")
)

hot_joined = hot_events.join(hot_dim, "join_key").drop("join_key", "salt")

# Cold path: untouched, broadcast the (still small) dimension
cold_joined = (
    events.where(~is_hot | is_hot.isNull())
          .join(F.broadcast(dim.where(~is_hot)), "customer_id")
)

result = hot_joined.unionByName(cold_joined)

That is more code than anyone wants in a pipeline, and I write it maybe twice a year. When it is the right tool, it is dramatic: a 90 GB single task becomes 64 tasks of about 1.4 GB, and the stage goes from 40 minutes to 90 seconds.

Two operational notes. The hot key list should come from a scheduled census, not a hardcoded constant that rots — I store it in a small config table and refresh it weekly. And SALT_N should be sized so hot-key rows divided by N lands near your advisory partition size, not at some round number like 100.

Fix four: make the join a broadcast

If the skewed side of the join has a small counterpart, broadcast it and the skew stops mattering. There is no exchange, so there is no partition to be lopsided. Every task processes its own local slice of the fact table regardless of key distribution.

# Skew is irrelevant when nothing is repartitioned by the key
result = events.join(F.broadcast(dim), "customer_id", "left")

This is why I check the broadcast option first for every skewed join before writing a line of salting code. Threshold tuning and when this backfires are covered in the shuffle post — the short version is that 64 MB is a saner default than Spark’s 10 MB, and you should know the dimension’s real size from table statistics.

Skew also shows up in aggregations, not just joins, and broadcast does not help there. For groupBy on a skewed key, use two-phase aggregation: group by key plus salt, then group by key again on the much smaller intermediate result.

-- Two-phase aggregation for a skewed group-by key
WITH partial AS (
  SELECT customer_id,
         pmod(hash(event_id), 32) AS salt,   -- deterministic spread
         count(*) AS c
  FROM prod.silver.events
  GROUP BY customer_id, pmod(hash(event_id), 32)
)
SELECT customer_id, sum(c) AS events
FROM partial
GROUP BY customer_id;

Pitfalls

Confusing skew with undersized clusters. If max task duration is 20x median, adding executors changes nothing — the critical path is one task. Read the distribution before you resize.

Salting everything. Uniform salting multiplies the small side by N and adds shuffle volume for keys that were never a problem. Salt the measured hot keys, keep the tail on the normal path.

Hardcoding a hot key list that rots. Last quarter’s whale customer churns, a new one appears, and your special-case code silently stops helping. Refresh the census on a schedule.

Assuming AQE covers you. It needs both the ratio and the absolute threshold to trigger, and it cannot split a single key. Verify isSkewJoin=true in the plan rather than assuming.

Writing skewed data to partitioned tables. A skewed key that is also a partition column produces one enormous file and hundreds of tiny ones, which is then a small-file and file-skipping problem in the transaction log.

Fixing skew with repartition on the same key. Repartitioning by the skewed column reshuffles the data into the same lopsided distribution, at full shuffle cost, for no benefit. It is the most common wrong first attempt.

FAQ

What ratio counts as skew?

Max task duration over median above 5 is my action threshold, and I cross-check with shuffle read bytes at the same percentiles. Under 3 is ordinary variance from executor placement and JVM warmup, not worth chasing.

Should I lower the AQE skew thresholds globally?

I run a factor of 3 and a 128 MB threshold on shared clusters and have not seen a regression. The cost of splitting a partition that did not strictly need it is small; the cost of missing real skew is a stage that runs 20x too long.

Does liquid clustering or Z-ordering fix skew?

No. Those change how data is laid out in files to improve skipping on reads. Skew happens at the shuffle boundary and is a function of key cardinality in the data itself, which no clustering scheme changes.

How do I pick the salt factor?

Divide the hot key’s row count by your target rows per task, which you can back out from the advisory partition size. If the hot key is 90 GB and you want 1.5 GB per task, 64 is right. Round numbers like 10 or 100 are usually too coarse or too fine.

Can I detect skew before the job runs?

Yes, and you should for critical pipelines. A daily job that runs the key census against yesterday’s partition and alerts when any single key exceeds a percentage of rows gives you a heads-up before the pipeline slows. Treat key distribution as part of the data contract.

What this means for your pipelines

Skew is a data property, so it deserves data monitoring rather than config tuning. For every large join in a scheduled pipeline, I know which columns are the join keys and roughly what their distribution looks like, and I have an alert for the top key’s share crossing a threshold. That turns skew from a 3 a.m. incident into a Tuesday ticket.

When it does happen, work the ladder in order: check for NULLs and sentinels first because that fix is nearly free and often reveals a real quality bug, then see whether a broadcast join removes the exchange entirely, then lower the AQE thresholds, and only then write salting code. I reach step four rarely, and when I do I keep the hot key list in configuration where it can be refreshed.

The larger habit is reading the task distribution instead of the job duration. Wall clock tells you something is wrong; the percentile table tells you what. Combined with the shuffle metrics and the executor memory picture, those three screens explain nearly every slow Spark job I have ever been handed.