Data skew

01 · What it is

One key too many, one task too slow

Data skew is rows spread unevenly across keys. After a shuffle — Spark moving rows across the network so every row with the same key lands in the same partition — a few "hot" keys pile into a few huge partitions while the rest stay tiny. The task that processes a huge partition runs for minutes after every other task has finished: a straggler. And a stage is only as fast as its slowest task.

analogy Skew is a supermarket where everyone happens to need the one cashier who handles loyalty cards — nine lanes empty, one lane 200 people deep. A bigger till for that lane doesn't help; opening more lanes for those customers does. That is the whole idea behind the two manual cures below.

There are exactly two levers here, and both come after the automatic one (AQE skew join, Lesson 05) has done what it can. The deep dive explains each once; for now, just hold the shape:

• Salting — add a small random suffix to the hot key so its rows spread across N partitions instead of one. Used for a skewed aggregation and a skewed join.
• SQL hints — /*+ … */ annotations right after SELECT that tell the planner what to do: which join strategy, or how to repartition.

02 · Why it matters

One straggler holds the whole cluster hostage

• The cluster sits idle waiting on one core. If 199 partitions hold 1 GB each and one holds 80 GB, 199 tasks finish in seconds and one runs for an hour — and may spill or OOM, because a single partition no longer fits in execution memory.
• It's the recurring villain of this track. Skew makes joins slow, causes spill and executor OOM (Lesson 04), and is exactly what AQE's skew-join split (Lesson 05) exists to fight. When AQE can't finish the job, salting is the manual lever.
• It's a constant interview probe: "Your join's last task runs 50× longer than the median — what's happening, and how do you fix it?" The expected answer is "skew — confirm AQE skew join, then salt the hot key," and you must be able to write the salted join. The how is the next section.

03 · How it works — deep dive

Spot it, then split the hot key

Every skew fix rests on one mechanic: the shuffle picks a row's partition from hash(key) % numPartitions, so equal keys collide into the same partition. If one key dominates, its partition dominates. Everything below follows from that. Each idea is explained here once — later sections show it, verify it, and help you choose, but don't re-explain it.

3.1 · First, see the skew in the Spark UI

Before any fix, confirm the straggler. Skew has an unmistakable signature:

• Stages tab → the slow stage → task table. The Max task duration is far above the Median — a 50×–500× gap is classic skew — and Max "Shuffle Read Size / Records" is far above the 75th percentile. A handful of tasks dominate the stage.
• SQL / DataFrame tab. The slow node is an aggregate or join right after an Exchange; per-task spill (memory)/(disk) shows up on the straggler.
• If Max ≈ Median, you don't have skew — look elsewhere. Salting always adds shuffle work, so it only pays off when one key truly dominates.

3.2 · Salting — the mechanism

use case A clickstream where ~90% of rows are country = 'US'. Group-by-country sends nearly all rows to one partition; salting splits US across 16 partitions.

• The move. Append a random integer 0..N-1 to the hot key, so the combined key (key, salt) hashes to N different partitions. The hot key's rows now spread across N tasks.
• Why it works. You trade "one task does 100% of a hot key" for "N tasks each do ~1/N of it." The skew flattens; the straggler disappears.
• The obligation. You changed the grouping key, so you must undo the salt afterwards — the second stage of a salted aggregation, or the explode step of a salted join. Salting also adds a shuffle, so use it only on genuinely skewed keys.

3.3 · Salting an aggregation — two-stage group-by

analogy Count a stadium by having each section count itself first (stage 1), then add the 50 section totals (stage 2) — instead of one person counting everyone.

• Stage 1 — partial aggregate on the salted key. Add salt = floor(rand()*N), group by (key, salt), compute a partial aggregate. The hot key is split into N groups across N partitions, so N tasks share the work.
• Stage 2 — final aggregate on the real key. Drop the salt and re-aggregate the N partials by key alone. Stage 1 already collapsed each (key, salt) to one row, so stage 2 is tiny.
• Composability. sum/count/min/max re-aggregate trivially. An average must be carried as sum/count partials and divided at the end — never average the averages.

3.4 · Salting a join — salt one side, explode the other

use case A 5 TB transactions fact (skewed on merchant_id, one mega-merchant = 40% of rows) ⋈ a small merchants dimension.

• The twist. If you salt only one side, salted rows won't find their match — MEGA#7 on the left has no MEGA#7 on the right, only MEGA.
• The fix. Salt the skewed (large) side's key with a random 0..N-1, and explode the other side so it holds one copy of each row per salt value (MEGA → MEGA#0 … MEGA#(N-1)). Join on the combined (key, salt).
• Why it works. The hot key's join work spreads across N partitions on both sides — no single task gets the whole hot key.

3.5 · SQL hints — the planner override

Hints are /*+ … */ comments placed immediately after SELECT that instruct the planner. Two families, each explained once here:

• Join-strategy hints — BROADCAST(t) (aliases BROADCASTJOIN, MAPJOIN), MERGE(t) (aliases SHUFFLE_MERGE, MERGEJOIN), SHUFFLE_HASH(t), SHUFFLE_REPLICATE_NL(t). Priority: BROADCAST > MERGE > SHUFFLE_HASH > SHUFFLE_REPLICATE_NL.
• Partitioning hints — COALESCE(n) (reduce partitions, no shuffle), REPARTITION(n | col | n,col) (full shuffle), REPARTITION_BY_RANGE([n,] col), REBALANCE([n][,col]) (even out partition sizes — ignored unless AQE is enabled).
• The trade-off. A hint overrides the cost-based optimizer, so a wrong hint can be worse than none (BROADCAST a 5 GB table → driver OOM). A scalpel for when you know the estimate is wrong.

Join-strategy priority: BROADCAST > MERGE > SHUFFLE_HASH > SHUFFLE_REPLICATE_NL.

3.6 · The order of operations (Rung 3)

• AQE skew join is the first line of defence (Lesson 05). It splits a skewed partition with no code change when the partition is > 5× the median and > 256 MB. Salt only what it can't split.
• The N× blow-up is the cost of a salted join. Exploding the whole other side ×N is wasteful for the 99% of keys that aren't skewed — so salt only the confirmed hot keys (give the rest a constant salt=0) and explode only for those keys.
• Pick N from the skew: roughly hot-key row count ÷ a healthy partition size. Too small → still skewed; too large → needless explosion and tiny partitions.
• REBALANCE is silently ignored unless AQE is on — a classic "why didn't my fix work?" trap.

Figure A · flatten the hot key

Watch the tall bar come down

One key (US, in red hatch) towers over the rest: a classic straggler. Drag the salt N slider to split that hot key across N partitions and watch the bars rebalance — and watch the join's small side get exploded ×N so every salted row still finds its match. This figure isolates one thing: the row distribution. Where it runs across the nodes is Figure B.

04 · The code — and how to verify it

Five moves, each with its proof

Every technique below is paired with how to prove it worked — the .explain() plan node or the Spark-UI signal. Apply, then verify; never assume.

Step 0 — confirm AQE skew join is on (before salting)

python# AQE skew join is the FIRST line of defence — it splits skewed partitions with no
# code change. Confirm it's on before reaching for manual salting.
spark.conf.get("spark.sql.adaptive.enabled")            # 'true' (default since Spark 3.2)
spark.conf.get("spark.sql.adaptive.skewJoin.enabled")   # 'true' (default)

# A partition is "skewed" if its size > skewedPartitionFactor (5.0) × median AND
# > skewedPartitionThresholdInBytes (256 MB); AQE then SPLITS it into sub-partitions.

# VERIFY — Spark UI SQL tab: the join's AQEShuffleRead node shows skewed=true, and the
# straggler's Max task time drops toward the Median after AQE splits it.

Salt a skewed aggregation (two-stage group-by)

pythonfrom pyspark.sql import functions as F

N = 16  # salt buckets — split each hot key across up to 16 partitions

# clicks is skewed: ~90% of rows are country = 'US'. A plain groupBy("country")
# sends nearly all rows to ONE partition -> one straggler task.

# Stage 1: partial aggregate on the SALTED key (key, salt). rand() spreads the hot
# key's rows across N buckets -> N partitions -> N tasks share the work.
salted = (clicks
    .withColumn("salt", (F.rand() * N).cast("int"))     # salt in 0..N-1
    .groupBy("country", "salt")
    .agg(F.sum("revenue").alias("partial_rev")))        # sum is re-aggregatable

# Stage 2: drop the salt, re-aggregate the N partials by the REAL key.
result = (salted
    .groupBy("country")
    .agg(F.sum("partial_rev").alias("revenue")))

# VERIFY — result.explain(mode="formatted") shows TWO HashAggregate/Exchange pairs
# (partial then final). Spark UI Stages tab: the partial-agg stage's Max task time is
# now close to its Median — the single straggler is gone.
result.explain(mode="formatted")

Composability note: use sum/count/min/max for the partial. For an average, carry sum and count partials and divide at the very end — averaging the stage-1 averages is wrong.

Salt a skewed join (build the salt key + explode the dimension)

pythonfrom pyspark.sql import functions as F

N = 16

# tx is skewed on merchant_id (one mega-merchant = 40%); merchants is small/unskewed.

# 1) Build the salt key on the SKEWED (large) side: append a random 0..N-1.
tx_salted = tx.withColumn("salt", (F.rand() * N).cast("int"))

# 2) EXPLODE the other side into N copies — one per salt value — so every salted
#    left row finds its matching right copy. explode(array(0..N-1)) = N rows per key.
merchants_exploded = (merchants
    .withColumn("salt", F.explode(F.array([F.lit(i) for i in range(N)]))))

# 3) Join on the COMBINED (merchant_id, salt) key.
joined = tx_salted.join(merchants_exploded, on=["merchant_id", "salt"], how="inner")

# VERIFY — joined.explain(mode="formatted") still shows a SortMergeJoin, but on
# (merchant_id, salt). Spark UI Stages tab: the join stage's Max task time now tracks
# the Median — the hot-key straggler is split across N tasks. Cost: merchants now has
# N× the rows (acceptable only because it's small).
joined.explain(mode="formatted")

Salt ONLY the hot keys (keep the blow-up proportional)

pythonfrom pyspark.sql import functions as F

N = 16
hot = ["MERCHANT_MEGA"]   # the few keys you confirmed are skewed (from the UI)

# Skewed keys get a real 0..N-1 salt; everyone else gets a constant 0 (no explosion).
tx_salted = tx.withColumn(
    "salt",
    F.when(F.col("merchant_id").isin(hot), (F.rand() * N).cast("int")).otherwise(F.lit(0)))

# Explode the other side ×N ONLY for hot keys; non-hot rows keep a single salt=0 copy.
m_hot  = (merchants.filter(F.col("merchant_id").isin(hot))
    .withColumn("salt", F.explode(F.array([F.lit(i) for i in range(N)]))))
m_cold = merchants.filter(~F.col("merchant_id").isin(hot)).withColumn("salt", F.lit(0))
merchants_salted = m_hot.unionByName(m_cold)

joined = tx_salted.join(merchants_salted, on=["merchant_id", "salt"], how="inner")

# VERIFY — same balanced task times, but merchants only grew N× for the few hot keys,
# not the whole dimension. Check Stages tab Max vs Median again.

Join-strategy & partitioning hints (DataFrame API + SQL)

pythonfrom pyspark.sql.functions import broadcast

# DataFrame-API equivalents of the SQL hints:
tx.join(broadcast(merchants), "merchant_id")             # == /*+ BROADCAST(merchants) */
tx.hint("merge").join(merchants, "merchant_id")         # == /*+ MERGE(...) */  force sort-merge
tx.hint("shuffle_hash").join(merchants, "merchant_id")  # == /*+ SHUFFLE_HASH(...) */
tx.repartition(200, "merchant_id")                     # == /*+ REPARTITION(200, merchant_id) */
tx.coalesce(50)                                       # == /*+ COALESCE(50) */

# VERIFY — .explain() shows the forced node (BroadcastHashJoin / SortMergeJoin /
# ShuffledHashJoin) or the Exchange with the new partition count / partitioning.

Equivalent Spark SQL — hints go immediately after SELECT:

sql-- Join-strategy hint (priority BROADCAST > MERGE > SHUFFLE_HASH > SHUFFLE_REPLICATE_NL).
SELECT /*+ BROADCAST(m) */ t.*, m.merchant_name
FROM   catalog.schema.transactions t
JOIN   catalog.schema.merchants m ON t.merchant_id = m.merchant_id;

-- Partitioning hints: repartition (full shuffle) or coalesce (no shuffle).
SELECT /*+ REPARTITION(200, customer_id) */ * FROM catalog.schema.transactions;
SELECT /*+ COALESCE(50) */                  * FROM catalog.schema.transactions;

-- REBALANCE evens out partition sizes — but is IGNORED unless AQE is enabled.
SELECT /*+ REBALANCE(merchant_id) */ * FROM catalog.schema.transactions;

Figure B · where the hot key lands

See it run across the cluster

This figure isolates a different thing from Figure A: not the distribution, but where the work physically lands. Step a skewed join across 1 driver + 3 executors and watch the hot key US pile onto one executor as a straggler — then flip to salted and watch it spread across all three. Use Prev / Play / Next.

05 · Decision table

Which lever, and what it costs

Read this by the symptom you're facing — it answers "which tool?" at a glance. (The why is §3; this is the which.)

06 · Edge cases & limitations

Where it gets subtle

New nuance an interviewer probes — not a re-run of §3:

• Non-composable aggregates. avg, countDistinct, percentile don't re-aggregate by naively re-applying the same function. Carry sum+count for avg; for distinct/percentile, two-stage salting is much harder — often AQE or a different approach wins.
• Skew AQE genuinely can't fix. AQE splits a skewed partition; if a single key is bigger than your split target, splitting partitions doesn't help — that's exactly when salting, which sub-divides the key itself, is required.
• A hint can hurt. BROADCAST on a table that's actually large → driver OOM / maxResultSize (Lesson 03); SHUFFLE_REPLICATE_NL on big inputs → a Cartesian-shaped explosion.

Hard limits worth stating plainly:

• Salting is manual and intrusive. You rewrite the query and must undo the salt correctly (re-aggregate / explode) — it adds at least one extra shuffle. Prefer AQE; salt the residual.
• Hints override the optimizer and AQE's runtime switch. A hint frozen into code can become wrong as data grows; hint only with a known reason.
• Databricks default. AQE (and its skew-join split) is on by default from DBR 7.3 LTS+; on OSS Spark it's default-on since 3.2. Databricks uses a higher runtime broadcast switch (30 MB) than the OSS 10 MB static threshold.

07 · Common mistakes

Gotchas — the mistake and its one-line fix

08 · References

Sources

• Apache Spark — SQL Performance Tuning (AQE skew join, coalesce, join strategies)
• Apache Spark — Join & partitioning hints (BROADCAST / MERGE / SHUFFLE_HASH / SHUFFLE_REPLICATE_NL, COALESCE / REPARTITION / REPARTITION_BY_RANGE / REBALANCE)
• Apache Spark — Configuration (spark.sql.adaptive.skewJoin.*, shuffle.partitions)
• Azure Databricks — Adaptive Query Execution (skew join, default-on DBR 7.3+)