Data skew in Apache Spark: how to see it, and six ways to fix it

Architecture: how a shuffle creates skew
The dataset
Step 1: measure it
Fix 1: broadcast the small side
Fix 2: let adaptive execution split the partition
Fix 3: salting
Fix 4: isolate the hot key
Fix 5: pre-aggregate before the join
Fix 6: change the key or the layout
Choosing between them
What failure looks like
Production notes
Frequently asked questions
Conclusion
References
Trademarks

TL;DR

Skew is not “slow”. It is one task doing most of the work while the rest of the cluster waits, so adding executors changes nothing.

Measure it before fixing it. The number that matters is the ratio of the largest partition to the median: on the dataset below it is 27x, and salting brings it to 1.9x.

Adaptive execution can split a skewed partition for you, but only when a partition exceeds 256 MB by default, so on smaller data it never fires however lopsided the key is.

Broadcasting the small side removes the shuffle entirely, which is the only fix that makes skew irrelevant rather than survivable. Reach for it first.

Every fix here changes the physical layout and must not change the answer. Each one below is checked against the unskewed result, because a salted join with a wrong replication step silently drops or duplicates rows.

A skewed job looks wrong in a specific way. The stage sits at 199 of 200 tasks complete for a long time, one executor’s GC climbs, and nothing you do to the cluster helps. That shape is diagnostic: the work is not too large, it is too unevenly divided, and a single task has become the critical path.

This post uses one deliberately skewed dataset throughout, measures the distribution, and then applies six fixes, reporting the partition distribution and a correctness check for each. The dataset is small so it runs on a laptop, which means the interesting numbers are structural, row counts and partition ratios, rather than timings. Timings on a laptop would tell you about page cache and JVM warm-up, not about skew.

Architecture: how a shuffle creates skew

Spark parallelises by partition, and a shuffle assigns rows to partitions by hashing the key. If one key holds most of the rows, one partition holds most of the rows, and one task processes them.

flowchart TB
  subgraph S["shuffle by city_id"]
    A["hash(city_id) % 8"]
  end
  A --> P0["partition 0<br/>5,000 rows"]
  A --> P3["<b>partition 3</b><br/><b>329,000 rows</b>"]
  A --> P5["partition 5<br/>8,000 rows"]
  A --> P7["partition 7<br/>12,000 rows"]
  P0 --> T0["task: done"]
  P3 --> T3["<b>task: still running</b>"]
  P5 --> T5["task: done"]
  P7 --> T7["task: done"]
  T3 --> R["the stage finishes<br/>when this one does"]

Two consequences follow, and both are counter-intuitive until you see the shape:

More executors do not help. The stage cannot finish before its longest task, and that task is one thread on one executor.
More partitions often do not help either. spark.sql.shuffle.partitions divides the key space, not a single key. Raising it from 200 to 2000 splits the small keys further and leaves the hot key exactly where it was.

That second point is why “increase shuffle partitions” is such common and such useless advice for skew. It is the right fix for partitions that are uniformly too big, and the wrong fix for one partition that is too big.

The dataset

One city holds 80% of the trips, which is the shape of most real skew: a default value, a null-substitute, a single huge tenant, or one popular product.

from pyspark.sql import SparkSession, functions as F

spark = (SparkSession.builder.appName("skew").master("local[4]")
         .config("spark.sql.shuffle.partitions", "8")
         .config("spark.sql.adaptive.enabled", "false")   # off, to see the raw problem
         .getOrCreate())

trips = (spark.range(0, 400000)
    .select(F.col("id").alias("trip_id"),
            F.when(F.col("id") % 100 < 80, F.lit(7))          # 80% land on city 7
             .otherwise(F.col("id") % 400).alias("city_id"),
            (F.col("id") % 4000 / 100.0).cast("double").alias("fare")))
trips.write.mode("overwrite").parquet("/tmp/skew/trips")

cities = spark.range(0, 400).selectExpr("id AS city_id", "concat('city-', id) AS city_name")
cities.write.mode("overwrite").parquet("/tmp/skew/cities")

t = spark.read.parquet("/tmp/skew/trips")
c = spark.read.parquet("/tmp/skew/cities")

Step 1: measure it

Two measurements are worth taking, and they answer different questions.

The key distribution tells you whether you have skew and which key is hot:

total = t.count()
top = t.groupBy("city_id").count().orderBy(F.desc("count")).limit(5).collect()
for r in top:
    print(f"city_id={r[0]:<5} rows={r[1]:<8} {100.0 * r[1] / total:5.1f}% of {total}")

city_id=7     rows=320000    80.0% of 400000
city_id=80    rows=1000       0.2% of 400000
city_id=82    rows=1000       0.2% of 400000
city_id=180   rows=1000       0.2% of 400000
city_id=96    rows=1000       0.2% of 400000

Every city other than 7 holds exactly 1,000 rows here, so which four appear after the first is an arbitrary tie-break and will differ on your run. The first line is the finding: one key out of 81 holds four fifths of the table.

The partition distribution tells you how bad it will be after the shuffle, which is the number that predicts the stall:

def partition_stats(df, label):
    rows = (df.withColumn("pid", F.spark_partition_id())
              .groupBy("pid").count().orderBy("pid").collect())
    counts = [r[1] for r in rows]
    median = sorted(counts)[len(counts) // 2]
    print(f"{label:<34} partitions={len(counts)} max={max(counts)} "
          f"median={median} ratio={max(counts) / median:.1f}")
    return counts

partition_stats(t.repartition(8, "city_id"), "repartitioned by city_id")

repartitioned by city_id           partitions=8 max=329000 median=12000 ratio=27.4

A ratio near 1 is healthy; this is 27. That single number is the one to track, because it is what changes when a fix works. It also survives being run on a sample, so you can measure it on a slice of production data without a full job.

In the Spark UI the same thing appears in the stage’s task table: sort by duration or by shuffle read size and look at max against the 75th percentile. The summary metrics row gives you min, 25th, median, 75th and max directly, and a max an order of magnitude above the median is the signature.

Fix 1: broadcast the small side

If one side fits in memory, broadcast it. There is then no shuffle of the large side, so there is no partition for the hot key to overfill and the skew stops mattering at all.

b = t.join(F.broadcast(c), "city_id")
ep = b._jdf.queryExecution().executedPlan().toString()

strategy = BroadcastHashJoin [city_id#1L], [city_id#3L], Inner, BuildRight
Exchange nodes in plan = 1
join rows = 400000

The one Exchange is the broadcast being built from the 400-row dimension, not a shuffle of the 400,000-row fact. Nothing about the key distribution matters any more.

This is the first thing to try and the most commonly missed, because the threshold is 10 MB and dimensions are often just over it as stored while being well under it after a filter and a projection. Reduce the small side first, then look at whether it broadcasts.

When it does not apply: both sides large, or a join type that forbids broadcasting that side. A full outer join can never be a broadcast hash join, and for a right outer only the left may be broadcast. Those rules are the subject of Apache Spark joins in depth.

Fix 2: let adaptive execution split the partition

Adaptive query execution can detect an oversized partition after the shuffle has run and split it into several, joining each piece against a copy of the matching rows from the other side. It is on by default.

spark.conf.set("spark.sql.adaptive.enabled", "true")
spark.conf.set("spark.sql.adaptive.skewJoin.enabled", "true")
j = t.join(c, "city_id").groupBy("city_name").agg(F.sum("fare").alias("s"))
qe = j._jdf.queryExecution()
j.collect()                       # execute this plan, then read it back
print(qe.executedPlan().toString())

# skewJoin.enabled = true
*(5) SortMergeJoin(skew=true) [city_id#1L], [city_id#6L], Inner
  +- AQEShuffleRead coalesced and skewed

# skewJoin.enabled = false
*(5) SortMergeJoin [city_id#1L], [city_id#6L], Inner
  +- AQEShuffleRead coalesced

(skew=true) on the join operator and and skewed on the shuffle read are the two markers that it fired.

The catch, and it is a big one. The rule computes one threshold as the larger of two numbers:

  def getSkewThreshold(medianSize: Long): Long = {
    conf.getConf(SQLConf.SKEW_JOIN_SKEWED_PARTITION_THRESHOLD).max(
      (medianSize * conf.getConf(SQLConf.SKEW_JOIN_SKEWED_PARTITION_FACTOR)).toLong)
  }

so a partition must exceed both skewJoin.skewedPartitionFactor times the median (default 5.0) and skewJoin.skewedPartitionThresholdInBytes (default 256MB). Taking the maximum is what makes the byte floor absolute: on a dataset where no partition reaches 256 MB the rule can never fire, however lopsided the distribution is. The example above only fires because I lowered that floor to 1MB.

So AQE skew handling is real and it is not a general answer. It helps on genuinely large partitions and does nothing on a 27x ratio between partitions of 300 MB and 12 MB. Check the markers rather than assuming.

Fix 3: salting

Salting is the general fix and the one to understand properly, because a half-implemented version silently changes results.

The idea: add a random bucket number to the key on the large side so the hot key becomes N keys, and replicate the small side once per bucket so every piece still finds its match.

flowchart TB
  subgraph L["large side"]
    A["city_id=7"] --> A1["(7, salt=0)"]
    A --> A2["(7, salt=1)"]
    A --> A3["(7, salt=...)"]
    A --> A4["(7, salt=7)"]
  end
  subgraph R["small side, replicated"]
    B["city_id=7"] --> B1["(7, 0)"]
    B --> B2["(7, 1)"]
    B --> B3["(7, ...)"]
    B --> B4["(7, 7)"]
  end
  A1 --> J["join on (city_id, salt)"]
  B1 --> J
  A4 --> J
  B4 --> J
  J --> O["same rows as the plain join,<br/>spread over 8 partitions"]

SALT = 8

# large side: one random bucket per row
tf = t.withColumn("salt", (F.rand(seed=7) * SALT).cast("int"))

# small side: one copy per bucket, so no match is lost
cf = c.withColumn("salt", F.explode(F.array([F.lit(i) for i in range(SALT)])))

salted = tf.join(cf, ["city_id", "salt"])

small side rows: 400 -> 3200 after explode by 8
fact repartitioned by (key, salt)   partitions=8 max=92841 median=50039 ratio=1.9
join rows = 400000   same as plain join: True

The ratio drops from 27.4 to 1.9, and the row count is unchanged. That second half is not a formality. The replication step is where salting goes wrong: forget the explode and rows of the hot key whose salt has no match disappear, and replicate the wrong side and you get duplicates. Always compare the row count, and ideally the full result, against the unsalted join.

Three practical notes:

The cost is the replication. The small side grows by a factor of SALT, so salting is cheap when it is small and pointless when it is not.
Salt only the hot keys if you can. Applying the salt conditionally, when(col("city_id") == 7, rand * SALT).otherwise(0), keeps the replication to one key and needs the dimension exploded only for that key.
Pick SALT from the ratio, not from taste. A 27x imbalance needs roughly 27 buckets to level; 8 got it to 1.9 here because the other partitions grew too.

Fix 4: isolate the hot key

When one key is hot and you know which, split the query: broadcast-join the hot key on its own, shuffle-join everything else, and union the two.

HOT = 7
hot = (t.filter(F.col("city_id") == HOT)
        .join(F.broadcast(c.filter(F.col("city_id") == HOT)), "city_id"))
cold = (t.filter(F.col("city_id") != HOT)
         .join(c.filter(F.col("city_id") != HOT), "city_id"))
result = hot.unionByName(cold)

hot rows=320000 cold rows=80000 total=400000
matches plain join: True

The hot side is a broadcast join, so it never shuffles. The cold side is a normal shuffle join over a now-uniform key distribution. It is more code than salting and it is easier to reason about, since nothing is replicated and nothing is random. It needs you to know the hot key, which the measurement step gives you.

Fix 5: pre-aggregate before the join

If the query aggregates after joining, do the aggregation first. Skew in an aggregation is far cheaper than skew in a join, because the hot key collapses to one row rather than producing a large intermediate.

# 400,000 rows into the join
join_then_agg = (t.join(F.broadcast(c), "city_id")
                  .groupBy("city_name").agg(F.sum("fare").alias("s")))

# 81 rows into the join
agg_then_join = (t.groupBy("city_id").agg(F.sum("fare").alias("s"))
                  .join(F.broadcast(c), "city_id"))

rows into the join: pre-aggregated=81 vs raw=400000
same answer: True

This is the largest structural win available when it applies, because it changes the size of the problem rather than its distribution. It applies only when the aggregate does not need columns from the dimension in its grouping, which is why the version above groups by city_id and joins for the name afterwards.

Fix 6: change the key or the layout

Two slower fixes, for skew you have to live with.

Bucketing. Writing both tables bucketed on the join key with the same bucket count removes the shuffle from every future join on that key. It does not fix the imbalance inside a bucket, so it pairs with salting rather than replacing it, but it takes the shuffle out of the repeated read path.

A composite key. If the hot key is hot because it is a placeholder, a -1, an UNKNOWN, or a null substitute, the real fix is upstream. Those rows often should not be joined at all. Filtering them out before the join, or replacing the placeholder with something genuinely distinct, is a data-modelling change that makes the skew disappear rather than spreading it.

Choosing between them

Situation	Fix	Why
Small side fits in memory after filtering	Broadcast	No shuffle, so skew is irrelevant
Query aggregates after the join	Pre-aggregate	Shrinks the input rather than spreading it
One or two known hot keys	Isolate them	No replication, no randomness, easy to verify
Many hot keys, or unknown ones	Salting	The general answer, at the cost of replicating the small side
Partitions genuinely over 256 MB	AQE skew join	Automatic, no query change
The same large join runs repeatedly	Bucketing, plus one of the above	Removes the shuffle permanently
The hot key is a placeholder	Fix it upstream	The rows usually should not be there

The order in that table is roughly the order to try things in. Broadcast and pre-aggregation change the shape of the work; salting and isolation only redistribute it; AQE only helps above its threshold.

What failure looks like

Skew rarely announces itself as skew. The symptoms, in the order they usually appear:

A stage stuck at “199/200 tasks complete”, sometimes for longer than the rest of the job took.
In the stage’s summary metrics, a max shuffle-read size an order of magnitude above the 75th percentile.
ExecutorLostFailure or a container killed by the resource manager, because the one big task exhausted its memory.
Repeated spill on one task only, visible as a large “Spill (disk)” for the max task and nothing for the median.
A job that got slower after you added executors, since the extra parallelism went to tasks that were already fast.

The last one is the clearest tell. If doubling the cluster changed nothing, the critical path is a single task, and no amount of hardware divides one partition.

Production notes

Measure before and after with the max-to-median partition ratio. It is one number, it is cheap to compute on a sample, and it is what tells you a fix worked.
Check the row count after any salting change. The replication step is easy to get subtly wrong, and the failure is silent: wrong results rather than an error.
Do not reach for spark.sql.shuffle.partitions for skew. It divides the key space, not a single key. It is the right knob for uniformly oversized partitions.
Do not assume AQE has it covered. Confirm (skew=true) in the final plan. The 256 MB default threshold means it often has not.
Salt only the hot keys where you can, so the small side is replicated for one key rather than all of them.
Look upstream when the hot key is a placeholder. A NULL-substitute key that joins to nothing useful is a modelling bug wearing a performance costume.
Re-measure after data growth. Skew is a property of the data, not the query, so a job that was balanced last quarter can stall on the same code.

Frequently asked questions

How much skew is too much? Watch the ratio of the largest partition to the median. Below about 2 it is not worth acting on. Around 10 the stage is noticeably waiting on one task. At 27, as here, the job is effectively serial in that stage.

Why did adding executors make no difference? Because the stage ends when its slowest task ends, and that task is one thread processing one partition. Extra executors give more parallelism to work that was already finishing early.

Does repartition fix skew? repartition(n) without a column fixes it, by shuffling rows round-robin, but it destroys the partitioning by key that the join needs, so the join then re-shuffles and the skew comes back. repartition(n, "key") reproduces exactly the distribution that is causing the problem.

Is salting always safe? Only if the replication matches the salting. The large side gets one random bucket per row and the small side gets a copy for every bucket. Get that wrong and rows vanish or duplicate with no error, which is why the row-count check belongs in the job and not just in the investigation.

Can I salt an aggregation instead of a join? Yes, and it is simpler: aggregate by (key, salt) first, then aggregate the partial results by key. No replication is needed because there is no second side, which is why skewed aggregations are much less trouble than skewed joins.

Does this apply to Hudi, Iceberg and Delta tables? The skew is in the shuffle, not the format, so all of it applies unchanged. The formats do help upstream: partitioning and clustering on the join key change what lands in each file, which changes the distribution before Spark ever shuffles it.

Conclusion

Skew is the one performance problem where the usual instincts are actively misleading. It does not respond to more executors, it does not respond to more shuffle partitions, and it does not look like a data problem until you measure the distribution.

The useful reframing is that every fix here does one of three things. It removes the shuffle, which is broadcasting. It shrinks what goes into the shuffle, which is pre-aggregation and filtering. Or it redistributes what the shuffle produces, which is salting, hot-key isolation and AQE’s skew split. The first two change the problem; the third divides it. Try them in that order, because a problem removed needs no tuning.

And whichever you pick, check the answer. Of the six fixes above, three replicate or split data, and all three can change results if the replication is wrong. The row-count comparison that took one line in every example here is the cheapest insurance in this entire area.

References

SQL performance tuning for the adaptive-execution and skew-join configuration defaults
OptimizeSkewedJoin.scala, the rule that performs the runtime split and the two thresholds it tests
ShufflePartitionsUtil.scala for how partitions are coalesced and split
Apache Spark joins in depth for the strategy and build-side rules that decide whether broadcasting is even available
Inside Spark’s Catalyst optimizer for where adaptive execution sits relative to the logical rewrites

Trademarks

Apache Spark, Apache Hudi, Apache Iceberg, Apache Parquet and Apache are either registered trademarks or trademarks of The Apache Software Foundation in the United States and other countries.