Data Skew Explained: Why One Task Runs Forever (and How to Fix It)

data skew is the reason a distributed job that should finish in four minutes instead crawls for four hours while 199 of its 200 tasks sat idle an hour ago — it is the single most common performance pathology in Spark, Flink, Trino, and every other shuffle-based engine, and it is invisible until you learn where to look. The wall-clock time of a distributed stage is not the average task time; it is the slowest task time, so the moment one partition ends up holding a hundred times more rows than the others, that one task becomes a straggler that the entire cluster waits on while every other executor core burns money doing nothing. The uneven work is not caused by a slow machine or a bad network — it is caused by the shape of your data, where one key, one region, one NULL, or one guest-checkout account carries a disproportionate share of the rows.

This guide is the walkthrough you wished existed the first time a job "hung at 99%," or an interviewer asked "why does one task run forever, and how would you fix it?", or a groupBy that used to be instant started spilling to disk and then died with an out-of-memory error. It works through what partition skew actually is and how to spot the straggler in the stage timeline, why joins amplify skew when a hot key floods a single reducer, how salting and key redistribution spread that hot key across many partitions, how aggregation and groupBy skew differ from join skew and demand a two-stage combine, and how engine features — Spark AQE, skew-join hints, repartition, and bucketing — detect and prevent skew before it detonates. Each section pairs a teaching block with a worked interview answer: code, a step-by-step trace, an output table, then a concept-by-concept breakdown of why the fix works.

When you want hands-on reps immediately after reading, drill the optimization practice library →, rehearse on the data-processing practice library →, and sharpen the grouping axis with the aggregation practice library →.

On this page

• What data skew is — spotting the straggler
• Skew in joins — hot keys and the task that runs forever
• Salting and key redistribution
• Aggregation and groupBy skew
• Engine features — Spark AQE, skew hints, and prevention
• Cheat sheet — data skew recipes
• Frequently asked questions
• Practice on PipeCode

1. What data skew is — spotting the straggler

Data skew is uneven work per partition, and the stage always waits on the slowest task

The one-sentence invariant: data skew is any distribution of rows across partitions where one (or a few) partitions hold vastly more data than the rest, so the task processing that partition runs far longer than its peers, and because a distributed stage cannot finish until its last task finishes, the whole job's wall-clock time collapses onto that single straggler regardless of how many cores are idle. Skew is not a slow disk, a slow node, or a slow network — those are hardware stragglers you fix with speculation. Data skew is a logical straggler baked into the shape of your keys, and no amount of extra executors will help, because the extra executors have nothing to do while the one overloaded task grinds through its outsized partition.

What skew actually is — the mental model.

• Partitions are the unit of parallelism. A Spark stage runs one task per partition. If you have 200 shuffle partitions and 200 cores, all 200 run in parallel — if the work is balanced.
• Shuffles redistribute rows by a key hash. After a groupBy, join, or repartition, each row lands in partition = hash(key) % numPartitions. Every row sharing a key lands in the same partition.
• One fat partition = one fat task. If key guest owns 90% of the rows, the partition holding guest holds 90% of the work, and its task runs roughly 0.9 / (1/200) = 180x longer than a perfectly-balanced task would.
• Wall clock = max(task time), not mean(task time). This is the whole story. A stage with 199 tasks at 2 seconds and one at 3 hours takes 3 hours.

The 2026 reality — every shuffle engine skews, and AQE only softens it.

• Spark is the poster child: sort-merge joins and groupBy both hash-partition by key, and a hot key produces a single monster task. Spark AQE (on by default since 3.2) auto-splits skewed join partitions but does nothing for skewed groupBy.
• Flink skews on keyBy; a hot key pins one subtask's slot at 100% while the rest idle, and backpressure propagates upstream.
• Trino / Presto / BigQuery / Snowflake all skew on join and aggregation redistribution; the cloud warehouses hide it behind autoscaling but you still pay for the slow query.
• The constant across all of them: skew is a property of the data, so the fix is almost always to reshape the key distribution (salting, pre-aggregation, broadcast) rather than to add hardware.

How to spot it — the Spark UI tells you in ten seconds.

• Stage page → task table → sort by Duration. If the max duration is 10-100x the median (75th percentile), you have skew. A healthy stage has max ≈ 1.5-2x median.
• Shuffle Read Size / Records column. The straggler task will show a shuffle-read size orders of magnitude larger than its peers — that is the fat partition arriving.
• Spill (Memory / Disk) columns non-zero on one task. The fat task can't fit its partition in memory, so it spills; heavy spill precedes the OOM that finally kills it.
• The event timeline. A wall of short green bars with one bar stretching off the right edge is the visual signature of a straggler.

What interviewers listen for.

• Do you say "the stage waits on the slowest task" and name the max-vs-median ratio? — required answer.
• Do you distinguish a data straggler (skew) from a hardware straggler (slow node) — the first is fixed by reshaping keys, the second by speculative execution? — senior signal.
• Do you propose measuring the skew factor (max partition / median partition) before reaching for a fix? — senior signal.
• Do you name the hot key as the root cause rather than "the cluster is slow"? — required answer.

Worked example — reading the Spark UI stage page

Detailed explanation. The fastest diagnosis of skew is the Spark UI's stage detail page. Before you touch any code, you open the stage that is hanging, sort the task table by duration, and compare the max against the median. The gap tells you whether you have skew and how bad it is. Walk through the interpretation for a stalled join stage.

• The symptom. The job UI shows a stage stuck at "199/200 tasks" for over an hour.
• The task table. 199 tasks completed in 1-3 seconds; task 47 is still RUNNING at 62 minutes.
• The shuffle-read column. Task 47 shows 48.2 GB shuffle read; the median task shows 240 MB.
• The spill column. Task 47 shows 30 GB spilled to disk; every other task shows 0.

Question. Given the stage-page numbers below, decide whether this is data skew, and estimate the skew factor.

Input.

Code.

# Programmatic version of "read the stage page":
# pull task metrics from the Spark REST API and compute the skew factor.
import requests

APP    = "application_1725500000000_0042"
STAGE  = 12          # the hanging stage id
ATTEMPT = 0

base = f"http://spark-history:18080/api/v1/applications/{APP}"
tasks = requests.get(f"{base}/stages/{STAGE}/{ATTEMPT}/taskList",
                     params={"length": 100000}).json()

read_sizes = sorted(t["taskMetrics"]["shuffleReadMetrics"]["remoteBytesRead"]
                    + t["taskMetrics"]["shuffleReadMetrics"]["localBytesRead"]
                    for t in tasks if t.get("taskMetrics"))

median = read_sizes[len(read_sizes) // 2]
mx     = read_sizes[-1]
skew_factor = mx / max(median, 1)

print(f"tasks={len(read_sizes)}  median_read={median/1e6:.0f} MB  "
      f"max_read={mx/1e9:.1f} GB  skew_factor={skew_factor:.0f}x")
# → tasks=200  median_read=240 MB  max_read=48.2 GB  skew_factor=201x

Step-by-step explanation.

1. The stall at "199/200" is the tell. When 199 tasks finish quickly and one runs on for an hour, the work is not balanced — the difference cannot be a slow node, because a slow node would drag a handful of tasks, not exactly one.
2. The shuffle-read column is the smoking gun. Task 47 pulled 48.2 GB while the median task pulled 240 MB — a 201x ratio. That 48.2 GB is a single fat partition arriving at one task, which is the definition of partition skew.
3. The non-zero disk spill on task 47 (and zero everywhere else) confirms it: only the fat task exceeds executor memory, so only it spills. Spill is slow (disk round-trips), which is why the fat task is now hours behind.
4. The skew factor — max partition ÷ median partition — is ~200x here. Any factor above ~5x is worth fixing; above ~50x it will usually spill and often OOM. This number is the input to every downstream decision (broadcast vs salt vs AQE).
5. The REST-API snippet automates the same read: it fetches every task's shuffle-read bytes, sorts, and divides max by median. Wiring this into a job's post-run hook turns "someone noticed the job was slow" into an automatic skew alert.

Output.

Rule of thumb. Diagnose skew before you fix it: open the stage page, sort tasks by duration, and divide the max shuffle-read size by the median. A ratio under 2x is healthy; over 10x means one hot partition; the size of the ratio picks the fix.

Worked example — quantifying the skew factor from the data itself

Detailed explanation. You don't have to wait for a job to hang to find skew — you can measure the key distribution directly. A single GROUP BY key COUNT(*) over the join or aggregation key, sorted descending, reveals the hot keys and lets you compute the skew factor offline. Every senior engineer runs this census before shipping a join on a new table.

• The census. SELECT key, COUNT(*) FROM t GROUP BY key ORDER BY 2 DESC — the top rows are your hot keys.
• The skew factor. max(count) / (total / distinct_keys) — how many times the hottest key exceeds a perfectly-even share.
• The decision threshold. Factor < 3 → ignore; 3-20 → salt the hot keys; > 20 with a small other side → broadcast; > 20 with two big sides → salt.

Question. Given an events table keyed by account_id, compute the skew factor and decide the fix.

Input.

Code.

-- PostgreSQL: measure key skew directly from the table
WITH counts AS (
    SELECT account_id, COUNT(*) AS n
    FROM   events
    GROUP  BY account_id
),
stats AS (
    SELECT MAX(n)                        AS max_n,
           SUM(n)                        AS total_n,
           COUNT(*)                      AS distinct_keys,
           SUM(n)::numeric / COUNT(*)    AS even_share
    FROM   counts
)
SELECT max_n,
       distinct_keys,
       ROUND(even_share)                 AS rows_if_even,
       ROUND(max_n / even_share, 1)      AS skew_factor
FROM   stats;

Step-by-step explanation.

1. The counts CTE is the census: one row per key with its row count. For a real table you would TABLESAMPLE or run it on a partition to keep the census cheap, but the shape is the same.
2. even_share = total_rows / distinct_keys is what each key would hold if the data were perfectly balanced. Here total ≈ 1.008 B rows over ~2 M keys ≈ 504 rows per key if even.
3. skew_factor = max_n / even_share = 900,000,000 / 504 ≈ 1.79 M. The guest account alone holds nearly 90% of all rows — a catastrophic skew that will pin one task forever.
4. The decision falls straight out of the numbers. Because the other side of any join to accounts is small (2 M account dim rows, easily a few hundred MB), the first move is a broadcast join. If both sides were large, salting the guest key would be the move (section 3).
5. Running this census in CI against a sample of every new large table is how teams stop shipping skewed joins. The hot key almost always has a business meaning — a sentinel 0/NULL, a default tenant, a bot account — and naming it early makes the fix obvious.

Output.

Rule of thumb. Measure skew from the data with a GROUP BY key COUNT(*) census and compute max_count / (total / distinct_keys). The hottest key almost always has a business meaning — a sentinel NULL, a default account, a bot — and naming it is half the fix.

Worked example — telling skew apart from a slow node

Detailed explanation. Not every straggler is data skew. A single slow disk, a noisy-neighbour VM, or a garbage-collection storm can also make one task lag. The diagnostic difference is decisive: skew is reproducible on the same partition and shows a huge shuffle read; a hardware straggler is random across runs and shows normal input size. Confusing the two sends you down the wrong fix path (speculation vs reshaping keys).

• Data skew. Same task index (same partition) is always the slow one; its shuffle-read size is huge; salting/broadcast fixes it.
• Hardware straggler. A random task is slow each run; its input size is normal; speculative execution re-runs it elsewhere and wins.
• The check. Compare the slow task's input/shuffle-read size against the median. Big → skew. Normal → hardware.

Question. Two jobs each have one slow task. Classify each and pick the fix.

Input.

Code.

# Classifier: skew vs hardware straggler, from task metrics
def classify_straggler(slow_task, median_read_bytes, median_dur_s):
    read_ratio = slow_task["shuffle_read"] / max(median_read_bytes, 1)
    dur_ratio  = slow_task["duration_s"]   / max(median_dur_s, 1)

    if read_ratio >= 5:
        return "DATA_SKEW: reshape the key (salt / broadcast / AQE)"
    if dur_ratio >= 5 and read_ratio < 2:
        # slow despite normal input → node/GC/disk problem
        return "HARDWARE_STRAGGLER: enable spark.speculation"
    return "BALANCED: look elsewhere (UDF cost, wide rows)"

print(classify_straggler({"shuffle_read": 48_200_000_000, "duration_s": 3720},
                         median_read_bytes=240_000_000, median_dur_s=2.1))
# → DATA_SKEW: reshape the key (salt / broadcast / AQE)

Step-by-step explanation.

1. Job A's slow task is always task 47 — the same partition — across three runs. Determinism points at the data: hash(hot_key) % 200 always lands on the same partition, so the same task is always the victim. Its 200x shuffle read confirms data skew.
2. Job B's slow task is a different task each run (12, 88, 3). Randomness rules out the data — a hot key would pin the same partition every time. Its shuffle read is normal (1.1x), and its GC time is elevated: this is a hardware/JVM straggler, not skew.
3. The classifier encodes the rule: a shuffle-read ratio ≥ 5x means skew (reshape the key); a duration ratio ≥ 5x with normal input means a hardware straggler (turn on speculation). Speculation re-runs a lagging task on another executor and takes whichever finishes first — useless for skew (the copy has the same fat partition) but a clean win for a bad node.
4. Turning on spark.speculation for Job A would waste resources: the speculative copy of task 47 gets the identical 48 GB partition and is equally slow. That is why "just enable speculation" is a wrong answer for skew.
5. The senior move is to run the classifier automatically and route the fix: skew → salting/broadcast/AQE; hardware → speculation + node draining. Naming which one you have is the entire diagnosis.

Output.

Rule of thumb. A straggler that is always the same task index with a huge shuffle read is data skew — reshape the key. A straggler that is a random task each run with normal input is a hardware/GC problem — enable speculation. Never fix skew with speculation; the speculative copy inherits the same fat partition.

Data engineering interview question on diagnosing stragglers

A senior interviewer often opens with: "A nightly Spark job that used to finish in six minutes now runs for over three hours. The Spark UI shows the last stage stuck at 999 of 1000 tasks complete, with one task still running. Walk me through how you'd confirm it's data skew rather than a slow node, quantify how bad it is, and identify the offending key — before you change any job code."

Solution Using a skew-factor diagnostic and per-partition row probe

# skew_probe.py — confirm skew, quantify it, and name the hot key
from pyspark.sql import SparkSession, functions as F

spark = SparkSession.builder.getOrCreate()
df = spark.table("analytics.events")          # the table feeding the slow stage
JOIN_KEY = "account_id"                        # the key the slow stage shuffles on

# 1. Per-key census — how many rows per key
census = (df.groupBy(JOIN_KEY)
            .count()
            .orderBy(F.desc("count")))

# 2. Skew statistics in one pass
stats = census.agg(
    F.max("count").alias("max_rows"),
    F.expr("percentile_approx(count, 0.5)").alias("median_rows"),
    F.sum("count").alias("total_rows"),
    F.count("*").alias("distinct_keys"),
).collect()[0]

even_share  = stats["total_rows"] / stats["distinct_keys"]
skew_factor = stats["max_rows"] / even_share

print(f"distinct_keys = {stats['distinct_keys']:,}")
print(f"rows_if_even  = {even_share:,.0f}")
print(f"max_rows      = {stats['max_rows']:,}  (skew_factor = {skew_factor:,.0f}x)")

# 3. Name the hot keys (top 5)
census.limit(5).show(truncate=False)

# 4. Prove it is the SAME partition every run (deterministic straggler)
NUM_SHUFFLE = int(spark.conf.get("spark.sql.shuffle.partitions", "200"))
hot_partition = (df.groupBy(JOIN_KEY).count()
                   .withColumn("partition",
                               F.pmod(F.hash(F.col(JOIN_KEY)), F.lit(NUM_SHUFFLE)))
                   .groupBy("partition")
                   .agg(F.sum("count").alias("rows_in_partition"))
                   .orderBy(F.desc("rows_in_partition")))
hot_partition.show(5, truncate=False)

Step-by-step trace.

1. Run the census (groupBy(key).count()) to get rows per key; this is cheap relative to the failing stage and needs no code change to the pipeline.
2. Collapse it to skew stats in one pass: max, median (percentile_approx), total, distinct. The median being tiny while the max is enormous is the numerical signature of skew.
3. Compute even_share = total / distinct and skew_factor = max / even_share. A factor in the thousands proves the straggler is logical, not hardware.
4. Name the hot keys with census.limit(5) — here account_id = 0, the guest-checkout sentinel. Now the straggler has a business explanation.
5. The partition probe reproduces Spark's own routing: pmod(hash(key), numPartitions) shows which shuffle partition each key maps to and confirms one partition absorbs ~90% of the rows — the same task index every run, which rules out a slow node.

Output:

Why this works — concept by concept:

• Per-key census — groupBy(key).count() is the ground-truth distribution of the shuffle key. Every skew fix depends on knowing which keys are hot and how hot; the census answers both without waiting for the job to fail again.
• Skew factor = max / even_share — normalising the hottest key against a perfectly-even share turns "it feels slow" into a number. That number is the input to the fix decision: broadcast, salt, or AQE.
• percentile_approx for the median — the median (not the mean) is the right baseline because the mean is dragged up by the hot key itself. percentile_approx computes it in one pass over billions of rows without a full sort.
• pmod(hash(key), n) partition probe — this is Spark's actual shuffle routing. Reproducing it offline proves the straggler is the same partition every run — the fingerprint of data skew versus a random hardware straggler.
• Cost — one shuffle for the census (O(rows), but a single cheap pass) plus a tiny aggregation. Compared to re-running the 3-hour failing job blind, the probe is minutes. It changes zero pipeline code and produces the hot key by name — the cheapest possible first move.

Optimization
Topic — optimization
Optimization problems on diagnosing stragglers and skew

Practice →

Practice →

2. Skew in joins — hot keys and the task that runs forever

A shuffle join hash-partitions by the join key, so one hot key floods a single reducer

The mental model in one line: a shuffle (sort-merge) join redistributes both sides so that every row sharing a join key lands in the same partition on the same task — which is exactly what makes the join correct, and exactly what makes a hot key catastrophic, because if one key owns most of the rows then one task receives most of the rows, spills, and runs forever while the rest of the cluster finishes in seconds. Join skew is the most common and most painful form of data skew, because joins are everywhere and the fat side is often a fact table with a natural sentinel key.

Why joins amplify skew.

• The shuffle is keyed by the join column. For fact JOIN dim ON fact.k = dim.k, both sides are repartitioned by k. All rows with k = guest go to one partition on each side, and their cross-product is computed on one task.
• Cardinality multiplies. If the fat key has 900 M rows on the fact side and 1 matching row on the dim side, one task emits 900 M rows — and that task holds 900 M input rows in memory to sort-merge.
• Spill then OOM. The fat task exceeds executor memory, spills sorted runs to disk, and if the partition is big enough, dies with OutOfMemoryError or Container killed by YARN for exceeding memory limits.
• The other 199 tasks are done. They processed their thin partitions in seconds and now sit idle, which is why the job appears "stuck at 99%."

The canonical symptom — one task runs forever.

• Stage stuck at N-1 of N tasks. The classic signature; the one remaining task is the fat one.
• Shuffle read for that task is 10-1000x the median. The fat partition arriving.
• Executor logs show spill and GC pressure on one host. Only the executor running the fat task struggles.
• Eventually: task failure, stage retry, then job failure. Retries re-run the same fat partition and fail again — retrying skew never helps.

Broadcast vs shuffle — the first fix to reach for.

• Broadcast (map-side) join. If one side fits in memory (default threshold ~10 MB, tunable to a few hundred MB), Spark ships the whole small side to every executor and joins locally — no shuffle, no skew. This is the single best fix when it applies.
• When broadcast applies. Star-schema joins: a huge fact table joined to a small dimension. The dimension broadcasts; skew disappears because there is no key-based shuffle.
• When broadcast does not apply. Two large tables (fact-to-fact), or a "small" side that is still too big to broadcast. Then you need salting (section 3) or AQE (section 5).
• The knob. spark.sql.autoBroadcastJoinThreshold and the broadcast() hint force it; setting it too high risks driver/executor OOM from an oversized broadcast.

What interviewers listen for.

• Do you explain that the join shuffles by the join key so a hot key concentrates on one task? — required answer.
• Do you reach for broadcast join first when one side is small? — required answer.
• Do you know broadcast doesn't help two large sides, and name salting/AQE for that case? — senior signal.
• Do you say retrying a skewed task never helps because it re-processes the same fat partition? — senior signal.

Worked example — reproducing a skewed join

Detailed explanation. To fix skew you first have to see it. The clearest reproduction is a fact table with a dominant sentinel key joined to a dimension. Build a clicks fact where 90% of rows carry user_id = 0 (logged-out / guest traffic) and join it to a users dimension. The result is a textbook one-task-runs-forever stage.

• Fact. clicks(click_id, user_id, ts, url) — 1 billion rows, 90% with user_id = 0.
• Dimension. users(user_id, plan, country) — 5 million rows.
• Join. clicks JOIN users ON clicks.user_id = users.user_id.
• Expectation. The partition for user_id = 0 receives ~900 M fact rows on one task.

Question. Write the naive join and predict which task becomes the straggler and why.

Input.

Code.

from pyspark.sql import SparkSession, functions as F

spark = (SparkSession.builder
         .config("spark.sql.shuffle.partitions", "1000")
         .config("spark.sql.adaptive.enabled", "false")   # disable AQE to SEE the skew
         .config("spark.sql.autoBroadcastJoinThreshold", "-1")  # force shuffle join
         .getOrCreate())

clicks = spark.table("web.clicks")     # 1B rows, user_id=0 is 90%
users  = spark.table("web.users")      # 5M rows

# Naive shuffle sort-merge join — will produce a single monster task
joined = clicks.join(users, on="user_id", how="inner")

joined.groupBy("plan").count().show()   # action that triggers the join

Step-by-step explanation.

1. Disabling AQE and setting autoBroadcastJoinThreshold = -1 forces a plain shuffle sort-merge join so the skew is visible rather than auto-handled — this is a teaching setup, not production config.
2. Spark repartitions both clicks and users by user_id into 1000 shuffle partitions. Every row with user_id = 0 hashes to the same partition — call it partition 512.
3. Partition 512 receives ~900 M click rows (from the fact) plus the single matching user_id = 0 dim row. The task for partition 512 must sort and merge 900 M rows; the other 999 tasks handle ~100 K rows each.
4. Task 512 exceeds executor memory, spills sorted runs to disk, and takes hours; the other 999 tasks finish in seconds. The stage sits at "999/1000" — the exact one-task-runs-forever picture.
5. Retrying does nothing: the retry of task 512 gets the same 900 M-row partition. The only fixes are to remove the shuffle (broadcast the 5 M-row users) or spread user_id = 0 (salting). Because users is small, broadcast is the right first move here.

Output.

Rule of thumb. A shuffle join concentrates each key on one task, so a fact table with a dominant sentinel key (0, -1, NULL, "guest") always produces a straggler. Identify the sentinel first; it is almost always the hot key.

Worked example — the broadcast-join fix

Detailed explanation. When one side of the join is small enough to fit in executor memory, broadcasting it eliminates the shuffle entirely — and with no key-based shuffle there is no skew, no matter how lopsided the fact side is. This is the first fix to try because it is the simplest and it fully removes the straggler.

• The move. Broadcast the small users dimension to every executor; each executor joins its slice of clicks locally.
• No shuffle of the fact. clicks stays partitioned as-is; each task probes the in-memory users map. The user_id = 0 rows are spread across whatever partitions they already sit in.
• The threshold. Raise autoBroadcastJoinThreshold or use the explicit broadcast() hint; ensure the driver and executors have headroom for the broadcast size.

Question. Rewrite the skewed join to broadcast the dimension, and explain why skew disappears.

Input.

Code.

from pyspark.sql import functions as F
from pyspark.sql.functions import broadcast

users_small = spark.table("web.users").select("user_id", "plan", "country")

# Broadcast the 180 MB dimension; no shuffle of the 1B-row fact
joined = clicks.join(broadcast(users_small), on="user_id", how="inner")

# Same downstream aggregation — now with no straggler
joined.groupBy("plan").count().show()

# Alternative: raise the auto-broadcast threshold so Spark chooses it itself
spark.conf.set("spark.sql.autoBroadcastJoinThreshold", str(256 * 1024 * 1024))  # 256 MB
# Spark now broadcasts any side whose estimated size <= 256 MB automatically.

Step-by-step explanation.

1. broadcast(users_small) tells Spark to collect the 5 M-row dimension to the driver, then ship one copy to every executor. At ~180 MB this is cheap and fits comfortably in executor memory.
2. With the dimension broadcast, the fact table is not shuffled. Each task reads its existing clicks partition and probes the in-memory users hash map by user_id. There is no repartition-by-key step, so there is no fat partition.
3. The user_id = 0 rows are still 90% of the data — but they stay spread across the fact's original partitions (by input file / block), which are balanced. Balanced input partitions plus a local map lookup means balanced tasks.
4. Selecting only the needed dimension columns (user_id, plan, country) shrinks the broadcast payload; broadcasting a wide dimension with 50 columns wastes memory and can push the driver toward OOM.
5. Raising autoBroadcastJoinThreshold to 256 MB lets Spark pick broadcast automatically whenever the estimated build side fits — but keep it below the point where a mis-estimated large side blows up the driver. The explicit broadcast() hint is safer when you know the small side is small.

Output.

Rule of thumb. If one side of a skewed join fits in memory (tens to a few hundred MB after column pruning), broadcast it and the skew vanishes because there is no key-based shuffle. Prune columns first to keep the broadcast small.

Worked example — the hot-key census before the join

Detailed explanation. Before you ship any join on a large table, run a hot-key census so you know whether skew is coming and can pick the fix in advance. The census is a one-time GROUP BY join_key COUNT(*) that surfaces the top keys and their share of the total. It costs one cheap pass and saves a 3-hour failure in production.

• The census query. Top-N keys by count, plus each key's percentage of the total.
• The threshold. Any single key above ~5% of the rows will skew a 200-partition shuffle; above ~50% it will spill.
• The output. A named hot key and a recommended fix (broadcast / salt / AQE).

Question. Run a hot-key census on clicks.user_id and translate the result into a fix decision.

Input.

Code.

-- PostgreSQL / Spark SQL: top hot keys and their share of the table
SELECT user_id,
       COUNT(*)                                            AS n,
       ROUND(100.0 * COUNT(*) / SUM(COUNT(*)) OVER (), 2)  AS pct_of_total
FROM   web.clicks
GROUP  BY user_id
ORDER  BY n DESC
LIMIT  10;

Step-by-step explanation.

1. The COUNT(*) per user_id is the row census; SUM(COUNT(*)) OVER () in the window computes the grand total in the same pass, so pct_of_total shows each key's share without a self-join.
2. Sorting descending and taking the top 10 surfaces the hot keys. Here user_id = 0 is 90.0% of the table — a single key owning nine-tenths of the rows guarantees a straggler on any key-based shuffle.
3. The percentage translates directly into a fix. A key above ~5% skews; user_id = 0 at 90% is a broadcast-or-salt situation. Because the users dim is small, broadcast wins; if both sides were large, salt the user_id = 0 rows only.
4. Keys 2 through 10 are all well under 1% — the "cold" keys. They never skew, which is why targeted salting (salt only the hot key) is enough and salting every key is wasteful (section 3).
5. Wiring this census into CI — fail the build if any single key exceeds a share threshold on a new large table — is how mature teams prevent skewed joins from ever reaching production.

Output.

Rule of thumb. Run a hot-key census (GROUP BY key COUNT(*), with each key's percentage of the total) before shipping any large-table join. One key above ~5% of rows will skew; the census names it and picks the fix before production does.

SQL interview question on a skewed join

A senior interviewer might ask: "You have a 2-billion-row transactions fact table joined to a 3-million-row merchants dimension. One merchant — an internal test merchant with merchant_id = 1 — accounts for 70% of all transactions, and the join stage runs for hours on a single task. The dimension is small. Walk me through detecting the hot key, the join rewrite you'd ship, and why it removes the straggler."

Solution Using a broadcast join with hot-key isolation

-- 1. Confirm the hot key with a census (Spark SQL / PostgreSQL)
SELECT merchant_id,
       COUNT(*)                                           AS n,
       ROUND(100.0 * COUNT(*) / SUM(COUNT(*)) OVER (), 1) AS pct
FROM   transactions
GROUP  BY merchant_id
ORDER  BY n DESC
LIMIT  5;
-- merchant_id = 1 -> ~70% of rows (the internal test merchant)

# 2. Primary fix: broadcast the small dimension (no key shuffle -> no skew)
from pyspark.sql.functions import broadcast

tx        = spark.table("fin.transactions")                       # 2B rows
merchants = spark.table("fin.merchants").select("merchant_id",    # 3M rows
                                                "merchant_name",
                                                "category")

result = tx.join(broadcast(merchants), on="merchant_id", how="inner")

# 3. Defensive fix if the dimension were NOT broadcastable:
#    isolate the hot key, broadcast just its one dim row, union the rest.
HOT = 1

tx_hot  = tx.filter(tx.merchant_id == HOT)
tx_cold = tx.filter(tx.merchant_id != HOT)

dim_hot  = merchants.filter(merchants.merchant_id == HOT)     # 1 row -> tiny broadcast
res_hot  = tx_hot.join(broadcast(dim_hot),  on="merchant_id")  # local, no shuffle
res_cold = tx_cold.join(merchants,          on="merchant_id")  # normal join, now balanced

result = res_hot.unionByName(res_cold)

Step-by-step trace.

1. Census confirms merchant_id = 1 owns ~70% of the fact — a guaranteed straggler on a shuffle join.
2. Broadcast the 3 M-row merchants dimension (only three needed columns) to every executor; the 2 B-row fact is not shuffled, so no partition concentrates the hot key.
3. Each task probes the in-memory merchant map locally; the 70% of rows on merchant_id = 1 stay spread across the fact's balanced input partitions.
4. The defensive variant (for when the dim can't broadcast) splits the fact into hot (= 1) and cold (!= 1): the hot side joins against a one-row broadcast dim (trivially small); the cold side does a normal, now-balanced shuffle join.
5. unionByName recombines the two result streams into one DataFrame with identical schema and correct results — the hot key handled without a shuffle, the cold keys handled normally.

Output:

Why this works — concept by concept:

• Broadcast (map-side) join — shipping the small dimension to every executor removes the key-based shuffle entirely. No shuffle means no partition can concentrate the hot key, so the straggler cannot form. This is the cleanest possible skew fix when one side is small.
• Hot-key isolation — when the dimension is too big to broadcast whole, filtering the fact into hot and cold streams lets you broadcast only the one dim row the hot key needs, while the cold keys take a normal balanced join. You broadcast bytes, not gigabytes.
• unionByName recombination — splitting and re-uniting is safe because the two branches produce the same schema and partition the input by key membership, so no row is dropped or duplicated. It is the standard shape for targeted skew fixes.
• Column pruning before broadcast — selecting only merchant_id, merchant_name, category keeps the broadcast small and the driver safe; broadcasting a wide dimension is how "broadcast join" itself becomes a driver-OOM incident.
• Cost — broadcast is O(dim) memory per executor and eliminates the O(fact) shuffle; hot-key isolation adds one extra scan of the fact (two cheap filters) but removes the O(fat-partition) straggler. Compared to the hours-long single task, both are minutes — the shuffle is the cost you delete.

Optimization
Topic — optimization
Optimization problems on skewed joins and broadcast

Practice →

Practice →

3. Salting and key redistribution

Salting appends a random suffix to the hot key so its rows spread across many partitions

The mental model in one line: salting is the technique where you append a small random integer suffix (the "salt", 0..N-1) to the skewed join or group key so that a single hot key becomes N synthetic sub-keys that hash to N different partitions — and to keep the join correct you replicate the other side across all N salt values, join on the composite (key, salt), then drop the salt — trading a bounded N-fold blow-up of the small side for the elimination of the straggler. Salting is the go-to fix when broadcast does not apply because both sides are large.

The salting recipe for a join.

• Salt the skewed (fact) side. Add salt = floor(rand() * N) to each fact row, and build the composite key (k, salt). The hot key's rows now spread across N partitions instead of one.
• Explode the other (dim) side. Each dim row must match all N salt values, so replicate every dim row N times with salt = 0..N-1. This is the cost: the dim grows N-fold.
• Join on the composite key. fact.(k, salt) = dim.(k, salt). Correct because every fact row's salt has a matching exploded dim row.
• Drop the salt. After the join, project away salt; results are identical to the un-salted join.

Choosing the salt factor N.

• The formula. N ≈ ceil(hot_key_rows / target_partition_rows), where target is what a healthy partition should hold (e.g. a few million rows or a few hundred MB).
• Too small. N under the skew factor leaves the hot key still concentrated — each sub-key still fat.
• Too large. N far above the skew factor explodes the dim side needlessly and creates tiny partitions with per-task overhead. Match N to the skew, not to a round number.
• Typical range. N of 8-256 covers most real skew; the guest/sentinel mega-key sometimes needs N in the low thousands.

Targeted (only-hot-key) salting — the senior refinement.

• The waste in naive salting. Salting every key explodes the dim N-fold across all keys, most of which are cold and never needed spreading.
• The refinement. Salt only the known hot keys; leave cold keys with a fixed salt (e.g. salt = 0), and explode the dim only for the hot keys. The dim grows by N × (hot key count) rows, not N × (all keys).
• How to know which are hot. The census from section 2. Hot keys are a short, stable list — usually a handful of sentinels.

What interviewers listen for.

• Do you remember to explode the other side so the join stays correct? — required answer (the most common thing candidates forget).
• Do you size N to the skew factor rather than picking an arbitrary number? — senior signal.
• Do you propose targeted salting (salt only hot keys) to avoid blowing up the whole dim? — senior signal.
• Do you drop the salt and confirm the result equals the un-salted join? — required answer.

Worked example — a full salted join in PySpark

Detailed explanation. The canonical salted join: two large tables where broadcast is impossible. Salt the fact's join key with N values, explode the dim by N, join on (user_id, salt), and drop the salt. This spreads the hot user_id = 0 across N partitions and removes the straggler.

• Fact. clicks (1 B rows, user_id = 0 is 90%).
• Dim. user_features — 200 M rows, too big to broadcast.
• Salt factor. N = 64 (spreads 900 M hot rows into ~14 M per sub-key).
• Result. Same rows as the naive join, no straggler.

Question. Write the full salted join and show the composite-key mechanics.

Input.

Code.

from pyspark.sql import functions as F

N = 64

# 1. Salt the fact side: each row gets a random salt in [0, N)
clicks_salted = (spark.table("web.clicks")
                 .withColumn("salt", (F.rand() * N).cast("int")))

# 2. Explode the dim side: every dim row is replicated for all N salts
dim = spark.table("web.user_features")
salts = spark.range(N).withColumnRenamed("id", "salt")     # 0..N-1
dim_exploded = dim.crossJoin(salts)                        # dim rows × N

# 3. Join on the COMPOSITE key (user_id, salt) — spreads the hot key
joined = clicks_salted.join(
    dim_exploded,
    on=["user_id", "salt"],
    how="inner",
)

# 4. Drop the salt — result is identical to the un-salted join
result = joined.drop("salt")
result.groupBy("plan").count().show()

Step-by-step explanation.

1. withColumn("salt", (rand()*N).cast("int")) gives every fact row a random salt in [0, 64). The 900 M rows with user_id = 0 now split into 64 groups of ~14 M, each with a distinct composite key (0, 0), (0, 1), … (0, 63).
2. Those 64 composite keys hash to (up to) 64 different shuffle partitions, so the former single fat partition becomes 64 balanced ones. The straggler is gone: no task holds more than ~14 M hot rows.
3. The dim must be exploded so each fact row finds its match: dim.crossJoin(salts) replicates every dim row 64 times, once per salt value. Now (user_id=0, salt=37) on the fact has a matching (user_id=0, salt=37) on the dim.
4. The join on ["user_id", "salt"] is correct because the salt was added to both sides consistently: a fact row with salt 37 only matches the dim replica with salt 37, and there is exactly one such replica per original dim row.
5. Dropping salt yields exactly the rows the naive join would have produced — salting changes only the physical partitioning, never the logical result. The cost is the 64x explosion of the dim (200 M → 12.8 B rows in the exploded intermediate), which is why targeted salting matters at scale.

Output.

Rule of thumb. Salting a join is a three-step contract: salt the skewed side, explode the other side by the same N, join on (key, salt), then drop the salt. Forgetting to explode the other side silently drops rows — it is the number-one salting bug.

Worked example — targeted salting of only the hot keys

Detailed explanation. Naive salting explodes the entire dimension N-fold even though only the hot keys need spreading. Targeted salting fixes that: salt (and explode) only the known hot keys, and leave the millions of cold keys with a fixed salt of 0. The dim grows by N × hot_key_count rows instead of N × all_keys — a huge saving when the dim is large.

• Hot list. From the census: hot = {0} (just the guest sentinel).
• Fact. Hot rows get a random salt 0..N-1; cold rows get salt 0.
• Dim. Hot dim rows explode ×N; cold dim rows get salt 0 only.
• Union. Both sides use the same composite-key join; no branching needed after salting.

Question. Rewrite the salted join to spread only user_id = 0, leaving cold keys untouched.

Input.

Code.

from pyspark.sql import functions as F

N = 64
HOT = [0]                        # from the census; the guest sentinel

# 1. Fact: hot rows get a random salt; cold rows get salt 0
clicks = spark.table("web.clicks")
clicks_salted = clicks.withColumn(
    "salt",
    F.when(F.col("user_id").isin(HOT), (F.rand() * N).cast("int"))
     .otherwise(F.lit(0)),
)

# 2. Dim: hot dim rows explode x N; cold dim rows keep salt 0
dim   = spark.table("web.user_features")
salts = spark.range(N).withColumnRenamed("id", "salt")

dim_hot  = (dim.filter(F.col("user_id").isin(HOT))
               .crossJoin(salts))                       # x N (few keys)
dim_cold = (dim.filter(~F.col("user_id").isin(HOT))
               .withColumn("salt", F.lit(0)))           # x 1

dim_salted = dim_hot.unionByName(dim_cold)

# 3. One composite-key join handles both classes
result = (clicks_salted
          .join(dim_salted, on=["user_id", "salt"], how="inner")
          .drop("salt"))

Step-by-step explanation.

1. The fact's salt is conditional: only rows whose user_id is in the hot list get a random salt; every cold row gets salt 0. Cold keys were never skewed, so they need no spreading.
2. The dim is built in two parts. dim_hot (only the hot keys) is exploded ×N; dim_cold (all the millions of cold keys) is tagged with salt 0 and not replicated. The union is the full salted dim.
3. Because there is exactly one hot key here, dim_hot is 1 × 64 = 64 rows and dim_cold is ~200 M rows — the dim barely grew. Naive salting would have produced 200 M × 64 = 12.8 B dim rows; targeted salting produces 200 M + 64.
4. A single join on (user_id, salt) handles both classes: hot fact rows (salt 0..63) match the 64 exploded hot dim rows; cold fact rows (salt 0) match the salt-0 cold dim rows. No post-join branching or union is needed on the result.
5. Dropping salt gives identical results to the naive join, at a fraction of the shuffle volume — the hot key is spread across 64 partitions while the cold keys are untouched. This is the production-grade salting pattern.

Output.

Rule of thumb. Salt only the hot keys the census names; give cold keys a fixed salt of 0 and do not explode their dim rows. Targeted salting keeps the dimension near its original size while still spreading the straggler.

Worked example — choosing the salt factor N

Detailed explanation. N is not a magic number — it is derived from the skew. Pick N so that the hottest key's rows, once divided by N, fit within a healthy partition size. Too small and the sub-keys are still fat; too large and you create per-task overhead and needlessly explode the dim.

• Target partition. Aim for a partition that fits in memory without spilling — say ~4 M rows or ~256 MB.
• The formula. N = ceil(hot_key_rows / target_partition_rows).
• Round up to a convenient bound. Powers of two (32, 64, 128) are fine; the exact value matters less than the order of magnitude.

Question. Compute N for the guest key and verify the resulting sub-key size.

Input.

Code.

import math

hot_key_rows        = 900_000_000
target_partition    = 4_000_000          # healthy partition size (rows)

N = math.ceil(hot_key_rows / target_partition)
N = 1 << (N - 1).bit_length()            # round up to next power of two (nicety)

per_subkey = hot_key_rows / N
print(f"N = {N}   rows_per_subkey = {per_subkey:,.0f}")
# → N = 256   rows_per_subkey = 3,515,625

# sanity: sub-key size should be <= the largest cold key, so no residual skew
print("balanced?" , per_subkey <= 4_200_000)   # → True

Step-by-step explanation.

1. Divide the hot key's 900 M rows by the target 4 M rows/partition: ceil(900M / 4M) = 225. Rounding up to the next power of two gives N = 256 (a convenience, not a requirement).
2. With N = 256, each hot sub-key holds 900M / 256 ≈ 3.52 M rows — below the 4 M target and below the largest cold key (4.2 M), so the former straggler is now no fatter than a normal partition.
3. The sanity check per_subkey <= max_cold_key is the correctness test for N: if the salted hot sub-keys are still larger than the biggest cold key, N is too small and skew remains. Here 3.52 M ≤ 4.2 M passes.
4. Picking N far larger (say 4096) would shrink sub-keys to ~220 K rows — pointlessly small, creating 4096 tiny tasks with scheduling overhead and exploding the hot dim rows 4096-fold. Match N to the skew, not to a bigger-is-safer instinct.
5. In production, recompute N from the census whenever the hot key's volume grows; a key that was 200 M rows last quarter and 900 M this quarter needs a larger N to stay balanced.

Output.

Rule of thumb. Size the salt factor as N = ceil(hot_key_rows / target_partition_rows), then verify each salted sub-key is no larger than your biggest cold key. N should track the skew factor — larger for hotter keys, not a fixed constant.

Data engineering interview question on salting

A senior interviewer might ask: "You must join two large tables — a 5-billion-row impressions fact and a 400-million-row campaigns dimension — on campaign_id. Neither fits in memory to broadcast, and one campaign (a house/default campaign, campaign_id = -1) carries 60% of impressions. The join runs for hours on one task. Design the salted join: how you'd salt, how you'd size N, how you keep it correct, and how you'd avoid exploding the whole 400 M-row dimension."

Solution Using a salted skew join with an exploded dimension

from pyspark.sql import functions as F
import math

# 1. Size N from the census: hot key = -1 with ~3B rows; target ~5M/partition
hot_key_rows     = 3_000_000_000
target_partition = 5_000_000
N = 1 << (math.ceil(hot_key_rows / target_partition) - 1).bit_length()   # -> 1024
HOT = [-1]

imp = spark.table("ads.impressions")       # 5B rows
cmp = spark.table("ads.campaigns")         # 400M rows

# 2. Salt the fact: hot key random salt, cold keys salt 0 (targeted)
imp_salted = imp.withColumn(
    "salt",
    F.when(F.col("campaign_id").isin(HOT), (F.rand() * N).cast("int"))
     .otherwise(F.lit(0)),
)

# 3. Explode only the hot dim rows x N; cold dim rows keep salt 0
salts = spark.range(N).withColumnRenamed("id", "salt")
cmp_hot  = cmp.filter(F.col("campaign_id").isin(HOT)).crossJoin(salts)     # 1 x 1024
cmp_cold = cmp.filter(~F.col("campaign_id").isin(HOT)).withColumn("salt", F.lit(0))
cmp_salted = cmp_hot.unionByName(cmp_cold)                                 # ~400M + 1024

# 4. Composite-key join, then drop the salt -> identical result, no straggler
result = (imp_salted
          .join(cmp_salted, on=["campaign_id", "salt"], how="inner")
          .drop("salt"))

result.groupBy("category").agg(F.count("*").alias("impressions")).show()

Step-by-step trace.

1. Size N from the census: the hot key -1 holds ~3 B rows (60% of 5 B); dividing by a 5 M target and rounding to a power of two gives N = 1024.
2. Salt the fact conditionally — only campaign_id = -1 rows get a random salt in [0, 1024); all cold campaigns get salt 0. The 3 B hot rows now split into 1024 sub-keys of ~2.9 M rows each.
3. Explode only the hot dim row ×1024 (1 row → 1024 rows) and tag the ~400 M cold dim rows with salt 0. The dimension grows by 1024 rows total — not by 1024×.
4. The composite-key join on (campaign_id, salt) matches hot fact rows to the 1024 exploded hot dim rows and cold fact rows to the salt-0 cold dim rows — one join, both classes, correct.
5. Drop the salt; the result equals the naive join row-for-row, but the former single fat partition is now 1024 balanced partitions of ~2.9 M rows, so no task runs forever.

Output:

Why this works — concept by concept:

• Salting the hot key — appending salt = rand()%N turns one hot key into N synthetic sub-keys that hash to N partitions, so the fat partition is split N ways. This is the core mechanism that dissolves the straggler.
• Exploding the other side — replicating each dim row across all N salt values preserves correctness: every salted fact row finds exactly one matching dim replica. Skip this step and salted rows silently fail to match — the classic salting bug.
• Targeted salting (hot-only) — salting only campaign_id = -1 means the dimension grows by 1024 rows, not 1024× (which would be 400 billion rows). The census makes this safe because the hot set is a tiny, known list.
• Sizing N to the skew factor — N = ceil(hot_rows / target) makes each sub-key no fatter than a healthy partition; a smaller N leaves residual skew, a larger N wastes tasks and dim explosion. N tracks the data, not a constant.
• Cost — one extra shuffle for the salted composite key, plus a negligible dim growth (1024 rows) under targeted salting. Compared to the hours-long single task, the salted join spreads the 3 B hot rows across 1024 balanced tasks — O(hot_rows / N) per task instead of O(hot_rows) on one.

Data processing
Topic — data-processing
Data-processing problems on salting and redistribution

Practice →

Practice →

4. Aggregation and groupBy skew

A skewed groupBy sends one group's rows to one reducer — fix it with a two-stage salted combine

The mental model in one line: aggregation skew is groupBy/reduceByKey skew — after the shuffle, all rows for one group land on one task, so a dominant group makes one reducer do most of the work — and the fix depends on the aggregate: algebraic aggregates (SUM, COUNT, MIN, MAX, AVG) can be partially aggregated map-side and then combined with a two-stage salted groupBy, while holistic aggregates (COUNT DISTINCT, MEDIAN) cannot be partially combined and need approximation or a different plan. Aggregation skew is subtler than join skew because Spark AQE does not auto-fix it — you own the fix.

Why groupBy skews.

• The shuffle concentrates each group. df.groupBy("country").sum("amount") repartitions by country; every row for country = US lands on one task. If US is 80% of the rows, one task processes 80% of them.
• The reducer holds the group's state. For a plain groupByKey, all raw rows for a group are shuffled to one task before aggregation — the fat group both floods the network and blows the reducer's memory.
• AQE won't save you. AQE's skew-join splitting applies to sort-merge joins, not to aggregation shuffles. A skewed groupBy needs an explicit fix.

Partial (map-side) aggregation — the free win for algebraic aggregates.

• What it is. For algebraic aggregates, each task pre-aggregates its local rows before the shuffle, so the shuffle carries partial results (one per key per partition) instead of raw rows. reduceByKey does this; groupByKey does not.
• Why it helps skew. The hot group still lands on one reducer, but that reducer now receives one partial per upstream partition (say 1000 partials) instead of 800 M raw rows — a massive reduction in shuffle and reducer memory.
• The DataFrame API does it automatically. df.groupBy(...).sum() inserts a partial aggregate; the raw-row hazard is mostly an RDD groupByKey footgun. Still, a single mega-group can leave the final combine lopsided, which is where two-stage salting comes in.

Two-stage salted aggregation — for when partial aggregation isn't enough.

• Stage 1 — salt and partial-aggregate. Add salt = rand()%N, group by (key, salt), and aggregate. The hot group splits into N partial groups on N tasks.
• Stage 2 — strip salt and combine. Group the N partials by the original key and combine them (SUM of partial SUMs, SUM of partial COUNTs). The final combine handles only N partials per key — tiny.
• Only for algebraic aggregates. SUM, COUNT, MIN, MAX combine trivially; AVG combines as sum(sums)/sum(counts). Holistic aggregates do not.

Holistic aggregates — COUNT DISTINCT and friends.

• The problem. COUNT DISTINCT can't combine partials: distinct(A) + distinct(B) ≠ distinct(A ∪ B) because of overlap. Two-stage salting silently over-counts.
• The fixes. (a) Approximate with HyperLogLog (approx_count_distinct), which is mergeable; (b) two-stage distinct — first distinct the (key, value) pairs (spreads the work), then COUNT; (c) accept a single reducer for the exact answer if the distinct set is small.

What interviewers listen for.

• Do you distinguish algebraic vs holistic aggregates and know which can two-stage combine? — senior signal.
• Do you know AQE does not fix groupBy skew — only join skew? — senior signal.
• Do you reach for approx_count_distinct (HLL) for skewed COUNT DISTINCT? — senior signal.
• Do you prefer reduceByKey over groupByKey (map-side combine) in RDD code? — required answer.

Worked example — two-stage salted SUM

Detailed explanation. The canonical aggregation-skew fix: a groupBy(country).sum(amount) where US dominates. Split into two stages — salt and partial-sum by (country, salt), then strip the salt and sum the partials by country. The hot group's work spreads across N tasks in stage 1; stage 2 combines only N partials per country.

• Skewed query. SELECT country, SUM(amount) FROM sales GROUP BY country; US = 80% of rows.
• Stage 1. Group by (country, salt) — US splits into N sub-groups.
• Stage 2. Group by country, SUM the partial sums.
• Correctness. SUM is algebraic: sum(partial sums) = total sum.

Question. Rewrite the skewed SUM as a two-stage salted aggregation.

Input.

Code.

from pyspark.sql import functions as F

N = 64
sales = spark.table("fin.sales")

# Stage 1: salt + partial aggregate by (country, salt)
stage1 = (sales
          .withColumn("salt", (F.rand() * N).cast("int"))
          .groupBy("country", "salt")
          .agg(F.sum("amount").alias("partial_sum")))

# Stage 2: strip salt, combine partials by country
result = (stage1
          .groupBy("country")
          .agg(F.sum("partial_sum").alias("total_amount")))

result.orderBy(F.desc("total_amount")).show()

# Contrast: the naive one-stage version that creates the straggler
naive = sales.groupBy("country").agg(F.sum("amount").alias("total_amount"))
# 'US' -> one reducer processes 800M rows' worth of partials on a single task

Step-by-step explanation.

1. Stage 1 adds a salt in [0, 64) and groups by (country, salt). The 800 M US rows split into 64 groups keyed (US, 0)…(US, 63), each ~12.5 M rows, landing on up to 64 different tasks. Each task computes a partial SUM for its sub-group.
2. Because SUM is algebraic, the partial sums are safe to combine later. Spark also applies map-side partial aggregation within stage 1, so the shuffle between the two groupBys carries only country × salt partials (a few thousand rows), not raw sales rows.
3. Stage 2 groups the partials by country alone and sums them: SUM(partial_sum) over the 64 US partials yields the exact total for US. The final combine touches only 64 rows per country — no straggler.
4. The naive one-stage version routes all 800 M US rows' aggregation to a single reducer; even with map-side partials, a single dominant group can leave the final task lopsided and slow. Two-stage salting removes that by spreading stage 1 across N tasks.
5. The result is identical to the naive aggregation because summing partial sums equals the total sum — salting changes the physical plan, not the arithmetic. The same shape works for COUNT (sum of partial counts) and AVG (sum of partial sums ÷ sum of partial counts).

Output.

Rule of thumb. For a skewed SUM/COUNT/MIN/MAX/AVG, use two stages: salt + partial-aggregate by (key, salt), then strip the salt and combine partials by key. The final combine sees only N partials per group, so the dominant group never lands whole on one reducer.

Worked example — reduceByKey vs groupByKey

Detailed explanation. In the RDD API, groupByKey shuffles every raw value to the reducer and only then aggregates — catastrophic under skew because the hot key's millions of raw values flood one task's memory. reduceByKey combines values map-side before the shuffle, so the hot key's task receives one partial per upstream partition. The choice is the single biggest RDD-level skew lever.

• groupByKey. Shuffles raw values; hot key → all raw values on one reducer → OOM.
• reduceByKey. Combines map-side; hot key → one partial per partition on the reducer → safe.
• Rule. Never groupByKey to then reduce; use reduceByKey/aggregateByKey.

Question. Show the two implementations of a per-key sum and explain the shuffle difference under skew.

Input.

Code.

rdd = spark.sparkContext.parallelize([])  # (key, amount) pairs, 1B of them

# BAD under skew: groupByKey shuffles all 800M raw 'hot' values to one reducer
bad = (rdd
       .groupByKey()                       # shuffle carries raw values
       .mapValues(lambda vals: sum(vals))) # reducer holds the whole iterable

# GOOD: reduceByKey combines map-side; reducer sees ~1 partial per partition
good = rdd.reduceByKey(lambda a, b: a + b) # shuffle carries partial sums

# aggregateByKey when the combine type differs from the value type
good2 = rdd.aggregateByKey(
    0,                                     # zero value
    lambda acc, v: acc + v,                # seqOp (map-side, within partition)
    lambda a, b: a + b,                    # combOp (across partitions)
)

Step-by-step explanation.

1. groupByKey performs no map-side combine: it shuffles every (hot, value) pair to the reducer holding hot. That reducer must materialise an iterable of 800 M values before sum runs — the fat group blows its memory and the task spills or OOMs.
2. reduceByKey applies the combine function a + b within each partition first, so each of the 1000 upstream partitions sends the reducer a single partial sum for hot. The reducer combines ~1000 partials instead of 800 M raw values — a 800,000x reduction in shuffled hot data.
3. The hot key still lands on one reducer with both approaches, but with reduceByKey that reducer's input is tiny (1000 partials), so there is no memory blow-up and no multi-hour task. Map-side combine is what makes the difference under skew.
4. aggregateByKey generalises reduceByKey when the accumulator type differs from the value type (e.g. building a (sum, count) tuple for AVG). It has the same map-side-combine benefit; the seqOp runs within a partition and the combOp across partitions.
5. In the DataFrame API you get map-side partial aggregation automatically, so this footgun is mostly an RDD concern — but knowing why reduceByKey beats groupByKey is a required signal that you understand where the shuffle volume comes from.

Output.

Rule of thumb. In RDD code, never groupByKey().mapValues(reduce) — use reduceByKey or aggregateByKey so values combine map-side before the shuffle. Under skew the difference is a working job versus an OOM on the hot key's reducer.

Worked example — COUNT DISTINCT skew with HyperLogLog

Detailed explanation. COUNT DISTINCT is holistic — you cannot combine two partial distinct counts because they may share values. Two-stage salting silently over-counts. The production fix is HyperLogLog (approx_count_distinct), whose sketches are mergeable, so the salted two-stage plan becomes correct (within a small error bound). When you need exactness, first distinct the (key, value) pairs to spread the work, then count.

• The trap. SUM(partial distinct counts) ≠ true distinct count (overlap double-counts).
• HLL fix. approx_count_distinct merges sketches exactly; error ~2%.
• Exact fix. Two-stage: DISTINCT (key, value) first (spreads), then COUNT per key.

Question. Compute distinct visitors per country when US dominates, without a single-reducer straggler.

Input.

Code.

from pyspark.sql import functions as F

visits = spark.table("web.visits")   # (country, visitor_id)

# APPROXIMATE (recommended for skew): HLL sketches merge correctly
approx = (visits
          .groupBy("country")
          .agg(F.approx_count_distinct("visitor_id", 0.02)   # ~2% error
                .alias("distinct_visitors")))

# EXACT, skew-safe: distinct the pairs first (spreads work), then count
exact = (visits
         .select("country", "visitor_id").distinct()          # heavy shuffle, but spread
         .groupBy("country")
         .agg(F.count("*").alias("distinct_visitors")))

# WRONG under skew: salting + SUM of partial distinct counts double-counts overlaps

Step-by-step explanation.

1. Plain countDistinct("visitor_id") grouped by country routes all 800 M US rows to one reducer, which must build a 40 M-element distinct set in memory — a classic aggregation straggler that spills and slows.
2. approx_count_distinct uses HyperLogLog: each partition builds a small fixed-size sketch of the visitors it saw, and sketches merge exactly (union of registers), so the hot group's work spreads across partitions and combines cheaply. The trade is ~2% error, tunable via the second argument.
3. HLL is the right default under skew because it sidesteps the holistic problem entirely — the sketch is mergeable where raw distinct counts are not. A 2% error on a 40 M distinct count is ±800 K, usually fine for dashboards and monitoring.
4. When exactness is required, the two-stage distinct plan works: SELECT DISTINCT country, visitor_id first deduplicates the (country, visitor_id) pairs — this shuffle is heavy but spread across the full pair space, not concentrated on country alone — then a cheap COUNT(*) per country counts the surviving rows.
5. The wrong plan — salt, COUNT(DISTINCT ...) per (country, salt), then SUM the partial distinct counts — double-counts any visitor who appears in multiple salt buckets, so US comes out inflated. Never SUM partial distinct counts.

Output.

Rule of thumb. COUNT DISTINCT is holistic — never combine partial distinct counts. Use approx_count_distinct (HLL sketches merge correctly) for skewed distinct counts, or a two-stage DISTINCT (key, value) then COUNT when you need the exact answer.

SQL interview question on aggregation skew

A senior interviewer might ask: "A daily SELECT country, SUM(revenue), COUNT(*) FROM orders GROUP BY country job has started spilling and running for hours. Profiling shows country = 'US' is 85% of the rows. Spark AQE is enabled but hasn't helped. Explain why AQE doesn't fix this, then design a skew-safe aggregation that computes the exact SUM and COUNT without a single-reducer straggler — and note how you'd handle a COUNT DISTINCT of buyers in the same query."

Solution Using two-stage salted aggregation

from pyspark.sql import functions as F

N = 128
orders = spark.table("fin.orders")     # (country, revenue, buyer_id), US = 85%

# Stage 1: salt + partial-aggregate the ALGEBRAIC parts by (country, salt)
stage1 = (orders
          .withColumn("salt", (F.rand() * N).cast("int"))
          .groupBy("country", "salt")
          .agg(F.sum("revenue").alias("partial_rev"),
               F.count("*").alias("partial_cnt")))

# Stage 2: strip salt, combine partials by country (SUM of sums, SUM of counts)
agg_exact = (stage1
             .groupBy("country")
             .agg(F.sum("partial_rev").alias("total_revenue"),
                  F.sum("partial_cnt").alias("order_count")))

# COUNT DISTINCT is holistic -> use HLL (mergeable), computed separately and joined
distinct_buyers = (orders
                   .groupBy("country")
                   .agg(F.approx_count_distinct("buyer_id", 0.02)
                         .alias("distinct_buyers")))

result = agg_exact.join(distinct_buyers, on="country", how="inner")
result.orderBy(F.desc("total_revenue")).show()

Step-by-step trace.

1. Why AQE doesn't help: AQE's skew handling splits oversized partitions in a sort-merge join; a groupBy aggregation shuffle is not a join, so AQE leaves the dominant US group on one reducer. The engineer owns this fix.
2. Stage 1 salts and partial-aggregates the algebraic measures. The 85% US rows split into 128 (US, salt) sub-groups across up to 128 tasks, each computing a partial SUM and COUNT.
3. Stage 2 combines the 128 partials per country: SUM(partial_rev) and SUM(partial_cnt) give the exact total revenue and order count — no reducer ever sees more than 128 partials per group.
4. COUNT DISTINCT of buyers cannot two-stage-combine (overlap), so it is computed with approx_count_distinct (HLL), whose sketches merge across partitions correctly, spreading the US work without a straggler.
5. The two results (algebraic exact, distinct approximate) join on country into one row per country — exact SUM/COUNT plus an approximate distinct-buyer count, all without a single-reducer straggler.

Output:

Why this works — concept by concept:

• AQE is join-only for skew — Adaptive Query Execution splits skewed partitions in sort-merge joins, not in aggregation shuffles. Recognising that a skewed groupBy is outside AQE's remit is why you must salt it yourself.
• Two-stage salted aggregation — stage 1 (groupBy (key, salt)) spreads the dominant group across N tasks; stage 2 (groupBy key) combines only N partials per group. For algebraic aggregates the arithmetic is exact.
• Algebraic vs holistic — SUM and COUNT combine trivially from partials; COUNT DISTINCT does not, because distinct sets overlap. Splitting the query into an exact algebraic part and an approximate distinct part is the correct decomposition.
• approx_count_distinct (HLL) — HyperLogLog sketches are mergeable, so the distinct count spreads across partitions and combines without a straggler, at ~2% error. It is the standard skew-safe distinct.
• Cost — two shuffles for the algebraic part (spread across N) plus one HLL pass, versus one shuffle that piles the dominant group on a single reducer. Each stage-1 task handles O(hot_rows / N); the final combine is O(N) per group — the straggler is gone at the price of one extra light shuffle.

Aggregation
Topic — aggregation
Aggregation problems on skewed group-by

Practice →

Practice →

5. Engine features — Spark AQE, skew hints, and prevention

Spark AQE splits skewed join partitions at runtime — but prevention beats every fix

The mental model in one line: Spark AQE (Adaptive Query Execution) uses runtime shuffle statistics to detect a partition that is far larger than the median and automatically split it into several balanced sub-partitions handled by separate tasks — which fixes skewed sort-merge joins with zero code change — but AQE does nothing for groupBy skew, so the durable answer is prevention: repartition on a good key, pre-aggregate, and bucket your tables so the skew never forms. Knowing exactly what AQE does and does not cover is the difference between "I turned on a flag" and "I understand my query plan."

What AQE does for skew.

• Runtime detection. After a shuffle, AQE reads the actual partition sizes. A partition qualifies as skewed if it is both larger than skewedPartitionFactor × median and larger than skewedPartitionThresholdInBytes.
• Automatic split. AQE splits the skewed partition into multiple sub-partitions (using the map-side statistics) and replicates the matching partition on the other join side, so each sub-task does a balanced slice of the join.
• Scope: sort-merge joins. AQE skew handling applies to SMJ (and shuffled-hash-join in recent versions). It does not apply to aggregation shuffles or window functions.
• Also. AQE coalesces many tiny partitions into fewer, and can switch a planned SMJ to a broadcast join when runtime stats show one side is small.

Enabling and tuning AQE.

• The flags. spark.sql.adaptive.enabled=true (default in 3.2+) and spark.sql.adaptive.skewJoin.enabled=true.
• The thresholds. spark.sql.adaptive.skewJoin.skewedPartitionFactor (default 5) and spark.sql.adaptive.skewJoin.skewedPartitionThresholdInBytes (default 256 MB). A partition must exceed both to be split.
• Advisory size. spark.sql.adaptive.advisoryPartitionSizeInBytes (default 64 MB) guides the target sub-partition size.

Skew-join hints — telling the optimizer what you know.

• The SKEW hint (Databricks / vendor Spark). SELECT /*+ SKEW('t', 'key') */ … tells the optimizer which column and values are skewed, so it applies the salting transform for you.
• broadcast() hint. Forces a broadcast join when you know a side is small, sidestepping skew entirely.
• When hints beat AQE. AQE reacts to runtime stats; a hint encodes what you already know from the census, avoiding a wasted first attempt.

Prevention — beating skew before it forms.

• Repartition on a better key. df.repartition(1000, "well_distributed_col") balances partitions when the natural key is skewed; repartition by a composite (key, salt) for a persistently hot key.
• Pre-aggregate early. Reduce rows before the expensive join/shuffle; a groupBy upstream shrinks a fat key's row count before it can skew a downstream join.
• Bucketing. bucketBy(n, "key") on write pre-shuffles the table by key so repeated joins skip the shuffle — but a skewed key still buckets unevenly, so bucket on a well-distributed column.
• Filter sentinels early. If user_id = 0 is meaningless to the join (guest traffic that has no user features), filter it before the join instead of salting it.

What interviewers listen for.

• Do you know AQE fixes join skew but not groupBy skew? — senior signal.
• Can you name the two thresholds (skewedPartitionFactor and skewedPartitionThresholdInBytes) a partition must exceed? — senior signal.
• Do you propose filtering the sentinel when the hot key carries no join value? — senior signal.
• Do you treat prevention (repartition, bucketing, pre-aggregate) as better than any runtime fix? — required answer.

Worked example — enabling and tuning AQE skew join

Detailed explanation. For a skewed sort-merge join, AQE can remove the straggler with no code change — you just configure it. Walk through enabling AQE skew handling and tuning the thresholds so a 48 GB fat partition gets split into balanced sub-partitions.

• Enable. AQE + skewJoin flags.
• Tune. Lower skewedPartitionFactor / threshold if a real skew isn't being split; raise them if AQE is over-splitting.
• Verify. The Spark UI SQL plan shows AQEShuffleRead with "skewed" sub-partitions.

Question. Configure AQE so a join whose fat partition is 48 GB (median 240 MB) gets split.

Input.

Code.

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

# A partition is 'skewed' if BOTH hold:
#   size > skewedPartitionFactor * median   (48 GB > 5 * 240 MB = 1.2 GB  ✓)
#   size > skewedPartitionThresholdInBytes  (48 GB > 256 MB              ✓)
spark.conf.set("spark.sql.adaptive.skewJoin.skewedPartitionFactor", "5")
spark.conf.set("spark.sql.adaptive.skewJoin.skewedPartitionThresholdInBytes", "256m")

# Target size for each split sub-partition
spark.conf.set("spark.sql.adaptive.advisoryPartitionSizeInBytes", "128m")

# No code change to the join itself:
joined = spark.table("web.clicks").join(spark.table("web.user_features"), on="user_id")
joined.groupBy("plan").count().show()

Step-by-step explanation.

1. Enabling adaptive.enabled and skewJoin.enabled turns on AQE's runtime skew handling for sort-merge joins. Both are defaults in Spark 3.2+, but setting them explicitly documents intent and covers older clusters.
2. AQE marks a partition skewed only if it exceeds both thresholds: > 5 × median (48 GB > 1.2 GB ✓) and > 256 MB (48 GB > 256 MB ✓). The double condition prevents splitting partitions that are merely a bit above median but still small.
3. The 48 GB fat partition is split into ~48 GB / 128 MB ≈ 375 sub-partitions of ~128 MB each (guided by advisoryPartitionSizeInBytes). AQE replicates the matching user_id = 0 row from the other side to each sub-partition so the join stays correct.
4. Each sub-partition is handled by its own task, so the former single 62-minute task becomes ~375 tasks of a few seconds each — the straggler is gone with zero change to the join code.
5. Verify in the Spark UI: the SQL tab's plan shows an AQEShuffleRead node annotated with "skewed" and the increased partition count. If a known skew isn't split, lower skewedPartitionFactor (e.g. to 2) or the byte threshold; if AQE over-splits small partitions, raise them.

Output.

Rule of thumb. For skewed sort-merge joins, enable adaptive.skewJoin and remember a partition must exceed both skewedPartitionFactor × median and skewedPartitionThresholdInBytes to be split. Lower the factor if a real skew isn't being handled; AQE fixes join skew with zero code change.

Worked example — repartition and bucketing to prevent skew

Detailed explanation. The most durable fix is to never form the fat partition. Repartitioning on a composite key spreads a hot key at read time, and bucketing a table on a well-distributed column pre-shuffles it so repeated joins skip the shuffle entirely. Prevention removes skew from every downstream query, not just one.

• Repartition. repartition(n, expr) to balance partitions; use a composite (key, salt) for a persistently hot key.
• Bucketing. bucketBy(n, "col") on write; joins/aggregations on col become shuffle-free — but bucket on a balanced column, or the buckets themselves skew.
• Pre-aggregate. Shrink the hot key upstream so it can't dominate a downstream join.

Question. Prevent the clicks-side skew before the join with repartition, and bucket the dimension for repeated joins.

Input.

Code.

from pyspark.sql import functions as F

N = 64

# 1. Repartition clicks on a composite (user_id, salt) so the hot key spreads
clicks_balanced = (spark.table("web.clicks")
                   .withColumn("salt", (F.rand() * N).cast("int"))
                   .repartition(1024, "user_id", "salt"))

# 2. Bucket the dimension once on write; future joins on user_id skip the shuffle
(spark.table("web.user_features")
 .write.mode("overwrite")
 .bucketBy(1024, "user_id")
 .sortBy("user_id")
 .saveAsTable("web.user_features_bucketed"))

# 3. Pre-aggregate clicks to shrink the hot key before any join
clicks_pre = (spark.table("web.clicks")
              .groupBy("user_id", "url")
              .agg(F.count("*").alias("hits")))   # collapses many rows per user

Step-by-step explanation.

1. Repartitioning clicks by (user_id, salt) into 1024 partitions spreads the hot user_id = 0 across partitions at read time, so the fat partition never forms before the join. This is salting applied as a repartition rather than as a join rewrite.
2. Bucketing user_features by user_id writes the table pre-shuffled into 1024 buckets. A later join of another bucketed table on user_id reads matching buckets locally with no shuffle — but note a skewed user_id still lands unevenly across buckets, so bucketing helps most when the join key is reasonably distributed.
3. Pre-aggregating clicks (grouping by (user_id, url) and counting) collapses many raw rows per user into one row per (user_id, url). If the downstream join only needs per-user hit counts, the hot key's row count drops by orders of magnitude before the join, dissolving the skew at the source.
4. These are complementary: repartition balances a one-off job, bucketing amortises the shuffle across many jobs, and pre-aggregation shrinks the data so no fix is needed. The best programs use pre-aggregation and bucketing so skew rarely reaches a runtime fix.
5. The trade-off is write-time cost (bucketing rewrites the table) and a fixed bucket count (changing it requires a rewrite). Choose the bucket count and column once, on a well-distributed key, for the join you run most.

Output.

Rule of thumb. Prevent skew rather than react to it: repartition on (key, salt) for a one-off job, bucket tables on a well-distributed join key for repeated joins, and pre-aggregate to shrink a hot key before it reaches a join. Prevention fixes every downstream query at once.

Worked example — when AQE isn't enough

Detailed explanation. AQE is not a universal skew solvent. It covers sort-merge join skew but leaves three common cases untouched: skewed groupBy aggregation, skewed window functions, and skew that causes an upstream OOM before AQE's runtime stats even apply. Knowing the gaps tells you when to fall back to salting.

• groupBy skew. AQE ignores aggregation shuffles — salt it (section 4).
• Window skew. PARTITION BY hot_key in a window concentrates one partition — repartition or pre-filter.
• Pre-shuffle OOM. If a map task OOMs before the shuffle (e.g. exploding a huge array for the hot key), AQE never gets to react — fix the map-side logic.

Question. For three skewed queries, decide whether AQE handles it or you must salt.

Input.

Code.

from pyspark.sql import functions as F, Window

# 1. Join skew -> AQE handles it (config only, section 5 example above)
j = fact.join(dim, on="key")           # AQE splits the fat join partition

# 2. groupBy skew -> AQE does NOT help; salt it (two-stage, section 4)
N = 64
g = (df.withColumn("salt", (F.rand()*N).cast("int"))
       .groupBy("key", "salt").agg(F.sum("v").alias("p"))
       .groupBy("key").agg(F.sum("p").alias("total")))

# 3. Window skew -> AQE does NOT help; reduce the partition or pre-filter
w = Window.partitionBy("key").orderBy("ts")
# If 'key' is hot, PARTITION BY key concentrates one task. Mitigations:
#  - pre-filter the sentinel if it doesn't need ranking
#  - narrow the window (add a sub-key to PARTITION BY where semantics allow)
ranked = df.filter(F.col("key") != 0).withColumn("rn", F.row_number().over(w))

Step-by-step explanation.

1. The join case is AQE's home turf: runtime stats reveal the fat partition and AQE splits it. No salting needed — just the config from the previous example.
2. The groupBy case is outside AQE. The aggregation shuffle concentrates the hot group on one reducer and AQE does not split aggregation partitions, so you apply the two-stage salted aggregation from section 4.
3. The window case is also outside AQE. PARTITION BY hot_key sends the entire hot partition to one task for ordering/ranking, and AQE's skew-join logic doesn't touch window shuffles. Mitigate by pre-filtering the sentinel (if it needs no ranking) or by narrowing the PARTITION BY with an extra sub-key where the ranking semantics allow.
4. The pre-shuffle OOM case is the subtlest: if a map-side operation (e.g. explode of a giant array attached to the hot key) blows memory before the shuffle, AQE's runtime stats never get a chance — you must fix the map-side logic (filter, limit, or restructure) rather than expecting any shuffle-time fix.
5. The decision rule: AQE for joins; salt for aggregations; repartition/pre-filter for windows; fix map-side logic for pre-shuffle OOM. "Enable AQE" is a correct answer only for the join case.

Output.

Rule of thumb. AQE fixes sort-merge join skew and nothing else. For skewed groupBy salt it, for skewed windows repartition or pre-filter, and for a map-side OOM fix the logic — "just enable AQE" only answers the join case.

Systems interview question on engine features and prevention

A senior interviewer might ask: "Your team enabled Spark AQE and some skewed jobs got faster, but a nightly aggregation and a windowed dedup are still stuck on one task. Explain what AQE actually did for the jobs that improved, why the aggregation and window jobs didn't benefit, and design a prevention-first plan — repartition, bucketing, pre-aggregation, sentinel filtering — so skew stops recurring across the whole pipeline."

Solution Using Spark AQE skew-join splitting plus repartition and bucketing

# 1. AQE handles the JOIN-skew jobs with zero code change
spark.conf.set("spark.sql.adaptive.enabled", "true")
spark.conf.set("spark.sql.adaptive.skewJoin.enabled", "true")
spark.conf.set("spark.sql.adaptive.skewJoin.skewedPartitionFactor", "5")
spark.conf.set("spark.sql.adaptive.skewJoin.skewedPartitionThresholdInBytes", "256m")
spark.conf.set("spark.sql.adaptive.advisoryPartitionSizeInBytes", "128m")

from pyspark.sql import functions as F, Window

# 2. The nightly aggregation was NOT helped by AQE -> two-stage salt it
N = 128
agg = (spark.table("fin.orders")
       .withColumn("salt", (F.rand() * N).cast("int"))
       .groupBy("country", "salt").agg(F.sum("revenue").alias("p"))
       .groupBy("country").agg(F.sum("p").alias("total_revenue")))

# 3. The windowed dedup was NOT helped by AQE -> pre-filter sentinel + repartition
w = Window.partitionBy("user_id").orderBy(F.desc("ts"))
dedup = (spark.table("web.events")
         .filter(F.col("user_id") != 0)                 # drop guest sentinel first
         .repartition(1024, "user_id")
         .withColumn("rn", F.row_number().over(w))
         .filter(F.col("rn") == 1))

# 4. Prevention: bucket the hot dimension once so repeated joins skip the shuffle
(spark.table("web.user_features")
 .write.mode("overwrite")
 .bucketBy(1024, "user_id").sortBy("user_id")
 .saveAsTable("web.user_features_bucketed"))

Step-by-step trace.

1. AQE split the fat partitions of the sort-merge join jobs at runtime — that is why those got faster with only the config flags, no code change.
2. The nightly aggregation didn't improve because AQE ignores aggregation shuffles; it is rewritten as a two-stage salted groupBy so the dominant country spreads across 128 sub-groups then combines.
3. The windowed dedup didn't improve because AQE ignores window shuffles; the guest sentinel (user_id = 0) is filtered out first (it needs no per-user dedup), and the remainder is repartitioned by user_id so the window's PARTITION BY lands on balanced tasks.
4. Bucketing user_features on user_id persists a pre-shuffled layout so every future join on user_id skips the shuffle entirely — prevention that pays back across the whole pipeline.
5. The combined plan routes each skew class to the right tool: AQE for joins, salt for aggregation, repartition/pre-filter for windows, bucketing for repeated joins — and skew stops recurring.

Output:

Why this works — concept by concept:

• AQE runtime skew split — AQE reads post-shuffle partition sizes and splits any partition exceeding both the factor and byte thresholds, replicating the other join side per sub-partition. It fixes sort-merge join skew with zero code change, which is why only those jobs improved.
• Two-stage salted aggregation — because AQE ignores aggregation shuffles, the dominant group must be spread by salting (key, salt) in stage 1 and combined in stage 2. This is the owner-supplied fix AQE cannot provide.
• Sentinel filtering — the guest user_id = 0 carries no per-user meaning for the dedup, so filtering it before the window removes the hot partition at the source — the cheapest possible fix when the hot key is semantically irrelevant.
• Bucketing on a stable key — writing the dimension pre-shuffled into fixed buckets makes repeated joins shuffle-free, amortising the cost across every future query instead of paying it each run.
• Cost — AQE adds a small runtime-planning overhead; salting adds one light shuffle; bucketing adds a one-time write cost and a fixed bucket count. Together they convert recurring multi-hour stragglers into balanced minutes-long stages — the shuffle and the straggler are the costs you delete, per job class.

Optimization
Topic — optimization
Optimization problems on AQE and skew prevention

Practice →

Data processing
Topic — data-processing
Data-processing problems on repartition and bucketing

Practice →

Cheat sheet — data skew recipes

• What skew is. Uneven rows per partition so one task (the straggler) runs far longer than its peers; the stage's wall clock equals the slowest task, not the average. Adding executors never helps — reshape the key distribution instead.
• Detection in the Spark UI. Open the stage page, sort tasks by duration, and divide the max shuffle-read size by the median. Under 2x is healthy; over 10x is one hot partition; the straggler also shows non-zero spill while every other task shows zero.
• Skew factor from the data. skew_factor = max(count) / (total_rows / distinct_keys) via SELECT key, COUNT(*) FROM t GROUP BY key ORDER BY 2 DESC. Any single key above ~5% of rows will skew a 200-partition shuffle; the hottest key almost always has a business meaning (sentinel 0/-1/NULL, default tenant, bot, guest).
• Skew vs hardware straggler. Same task index every run + huge shuffle read = data skew (reshape the key). Random task each run + normal input = hardware/GC straggler (enable spark.speculation). Never fix skew with speculation — the speculative copy inherits the same fat partition.
• Skew-join fix order. (1) Broadcast the small side (broadcast(dim) or raise autoBroadcastJoinThreshold) — no shuffle, no skew. (2) If both sides are large, salt. (3) If it's a sort-merge join, let AQE split it. (4) Filter the sentinel if it carries no join value.
• Salting a join. Salt the skewed side (salt = rand()%N), explode the other side ×N (crossJoin(range(N))), join on (key, salt), drop the salt. Forgetting to explode the other side silently drops rows — the number-one salting bug.
• Salt factor N. N = ceil(hot_key_rows / target_partition_rows), then verify each salted sub-key ≤ your biggest cold key. Typical N is 8-256; a sentinel mega-key sometimes needs the low thousands. Round to a power of two for convenience.
• Targeted salting. Salt only the census-named hot keys; give cold keys a fixed salt of 0 and don't explode their dim rows. Keeps the dimension near its original size (grows by N × hot_key_count, not N × all_keys).
• groupBy skew (two-stage). Stage 1: groupBy(key, salt).agg(partial); stage 2: groupBy(key).agg(combine partials). Works for algebraic aggregates (SUM, COUNT, MIN, MAX, AVG as sum/count). AQE does not fix aggregation skew — you own it.
• COUNT DISTINCT skew. Holistic — never SUM partial distinct counts (overlap double-counts). Use approx_count_distinct (HLL sketches merge correctly, ~2% error) or a two-stage DISTINCT (key, value) then COUNT for exactness.
• RDD rule. reduceByKey/aggregateByKey (map-side combine) over groupByKey (shuffles raw values). Under skew that's a working job versus an OOM on the hot key's reducer.
• AQE knobs. spark.sql.adaptive.enabled=true, spark.sql.adaptive.skewJoin.enabled=true; a partition is split only if it exceeds both skewedPartitionFactor × median (default 5) and skewedPartitionThresholdInBytes (default 256 MB). Applies to sort-merge joins only.
• Prevention. Repartition on (key, salt), bucket tables on a well-distributed join key (bucketBy), pre-aggregate to shrink a hot key before a join, and filter sentinels that carry no join value. Prevention fixes every downstream query at once.

Frequently asked questions

What is data skew in one sentence?

Data skew is an uneven distribution of rows across the partitions of a distributed job, so one partition holds far more data than the others and the task processing it — the straggler — runs much longer than its peers. Because a distributed stage cannot finish until its slowest task finishes, the whole job's wall-clock time collapses onto that one overloaded task while the rest of the cluster sits idle. Skew is a property of the data (a hot key, a sentinel NULL, a dominant region), not of the hardware, which is why the fix is to reshape the key distribution — broadcast, salting, AQE, or pre-aggregation — rather than to add more executors.

Why does one Spark task run forever while the others finish?

Because a shuffle routes every row with the same key to the same partition on the same task, so if one key owns most of the rows, one task receives most of the rows. That task has to sort, join, or aggregate a partition that is tens or hundreds of times larger than the median, so it exceeds executor memory, spills sorted runs to disk, and grinds for minutes or hours while the other 199 tasks finished their thin partitions in seconds. This is why the job appears "stuck at 99%": 199 of 200 tasks are done and the stage is waiting on the single fat one. Retrying does not help — the retry gets the identical fat partition — so the fix is always to reshape the key (broadcast the small side, salt the hot key, or let Spark AQE split the partition).

What is salting and when do I use it?

Salting appends a small random integer suffix (0..N-1) to a skewed key so that one hot key becomes N synthetic sub-keys that hash to N different partitions, spreading the straggler's work across N tasks. You use it when a join or groupBy has a hot key and you cannot broadcast the other side because both are large. For a join you must also replicate ("explode") the other side across all N salt values so every salted row still finds its match, then join on the composite (key, salt) and drop the salt; forgetting to explode the other side silently drops rows. Size N as ceil(hot_key_rows / target_partition_rows) and prefer targeted salting — salt only the census-named hot keys and leave the cold keys with a fixed salt of 0 — so the dimension barely grows.

Does Spark AQE fix all data skew?

No. Spark AQE (Adaptive Query Execution) detects an oversized shuffle partition at runtime and automatically splits it into balanced sub-partitions, but only for sort-merge joins (and shuffled-hash joins in recent versions). It does nothing for skewed groupBy/aggregation shuffles, skewed window functions (PARTITION BY hot_key), or a map-side out-of-memory error that happens before the shuffle. So "enable AQE" is a correct answer only for join skew; a skewed aggregation still needs a two-stage salted groupBy, a skewed window needs a repartition or a sentinel pre-filter, and a map-side OOM needs a logic fix. A partition is also only split if it exceeds both skewedPartitionFactor × median and skewedPartitionThresholdInBytes, so a genuinely skewed but small partition may be left alone.

How do I fix a skewed groupBy or aggregation?

Use a two-stage salted aggregation, and only for algebraic aggregates. Stage one adds salt = rand()%N and groups by (key, salt) so the dominant group splits into N partial groups on N tasks; stage two groups by the original key and combines the partials — SUM of partial sums, SUM of partial counts, MIN/MAX of partials, and AVG as sum(sums)/sum(counts). This works because those aggregates are algebraic (combinable from partials). COUNT DISTINCT is holistic and cannot be combined from partials, so use approx_count_distinct (HyperLogLog sketches merge correctly, ~2% error) or a two-stage DISTINCT (key, value) then COUNT when you need an exact answer. In RDD code, always prefer reduceByKey/aggregateByKey over groupByKey so values combine map-side before the shuffle.

How do I detect data skew before a job fails?

Run a hot-key census on the shuffle key — SELECT key, COUNT(*) AS n, 100.0*COUNT(*)/SUM(COUNT(*)) OVER () AS pct FROM t GROUP BY key ORDER BY n DESC LIMIT 10 — and compute the skew factor max(n) / (total / distinct_keys). Any single key above ~5% of rows will skew a typical 200-partition shuffle, and the top of the list usually names the culprit (a sentinel 0/-1/NULL, a default account, a bot, guest traffic). After a job runs, confirm skew in the Spark UI by sorting the stage's tasks by duration and dividing the max shuffle-read size by the median — over ~10x means one fat partition, and the straggler will show disk spill while every other task shows none. Wiring the census into CI (fail the build if any key exceeds a share threshold on a new large table) stops skewed joins from ever reaching production.

Practice on PipeCode

• Drill the optimization practice library → for the straggler-diagnosis, skew-join, broadcast, and AQE-tuning problems senior interviewers love.
• Rehearse on the data-processing practice library → for the salting, repartition, bucketing, and key-redistribution patterns that reshape a hot key.
• Sharpen the grouping axis with the aggregation practice library → for two-stage salted group-by, algebraic-vs-holistic aggregates, and approximate distinct counts.
• Stack the prerequisites against PipeCode's broader 450+ data-engineering catalogue to anchor the detect → reshape → prevent skew workflow against real graded inputs.

Lock in data-skew muscle memory

Docs explain the concept. PipeCode drills explain the decision — when to broadcast versus salt, why AQE fixes joins but not group-bys, how to size the salt factor, and how to spot the straggler in the Spark UI before it becomes a 3 AM page. Pipecode.ai is Leetcode for Data Engineering — pattern-first practice tuned for the production trade-offs data engineers actually face.

Practice optimization problems →
Practice data-processing problems →