🔥 Spark & Databricks · Chapter 6

Handling skew: salting and AQE

199 tasks finish in 20 seconds. One runs for 70 minutes. Adding executors will change nothing.

Study this lesson free in the app →

The basics

Data skew is when one key holds far more rows than the others, so after a shuffle one partition — and therefore one task — does most of the work.

Why it matters: a stage finishes only when its slowest task finishes. If driver_id = 'UNASSIGNED' is 30% of Meghalaya Cabs' rides, hash partitioning sends all of those rows to one partition. One core grinds for an hour while the rest of the cluster sits idle, and often that one task spills to disk or runs out of memory.

Two families of fixes:

• AQE (Adaptive Query Execution) — on by default since Spark 3.2. After a shuffle it looks at real partition sizes and, for sort-merge joins, splits oversized partitions automatically.
• Salting — you change the key yourself: add a random number to the hot key so its rows spread across N partitions, then undo it afterwards.

Mental model: one checkout lane for every customer whose surname starts with S. You either open extra S lanes on the fly (AQE), or hand S customers a numbered token that sends them to one of five lanes (salting).

Explained

Meghalaya Cabs' nightly job joins 900M ride requests to driver profiles. Every ride cancelled before a driver accepted has driver_id = 'UNASSIGNED' — 270M rows, 30% of the table.

The Spark UI: 1,999 tasks done in under a minute, task 1,412 at 74 minutes and counting, 48GB spilled.

The team doubled the cluster. Task 1,412 took 73 minutes.

The S lane

A stadium has 26 entry gates, one per surname initial. Gates Q, X and Z are empty. Gate S has a queue past the parking lot.

Hiring more guards for other gates does nothing. Two fixes work:

Open extra S gates when you see the queue — the supervisor watches the lines and splits S across three gates on the spot. That is AQE.

Give every S ticket a number 1–5 printed at booking, routing them to gates S1–S5. It works even for a crowd the supervisor cannot see coming — but the five gates' head counts must be added up at the end. That is salting.

What skew looks like in the Spark UI

Stage 7 — SortMergeJoin, 2000 tasks
Summary Metrics       Min     25th    Median   75th     Max
Duration              8 s     12 s    14 s     19 s     74 min
Shuffle Read Size     41 MB   55 MB   60 MB    68 MB    46.2 GB
Spill (Disk)          0       0       0        0        48 GB

Timeline:
exec 1 ▇▇▇ ▇▇ ▇▇▇ ▇▇ ...   done
exec 2 ▇▇ ▇▇▇ ▇▇ ▇▇▇ ...   done
...
exec 7 ▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇ ← task 1412

Max ≫ 75th percentile is the signature. If all percentiles are high together, it is volume, not skew.

What AQE does — and doesn't

AQE re-optimises at each shuffle boundary using real sizes from the finished map stage. Three features, all default-on in Spark 3.2+:

1. Coalesce shuffle partitions — merges tiny post-shuffle partitions toward advisoryPartitionSizeInBytes (64MB), so 200 or 2,000 partitions become a sensible number
2. Switch join strategy — if one side turns out small enough, a planned sort-merge join becomes a broadcast join
3. Skew join — a partition is skewed if it is > skewedPartitionFactor (5) × median and > skewedPartitionThresholdInBytes (256MB). AQE splits it into several tasks and replicates the matching partition from the other side to each

Limits: skew splitting applies to sort-merge joins (and shuffled hash joins in newer versions). In an outer join only the preserved side can be split (the left side of a left join), and full outer joins are not handled. It does not split skewed groups in groupBy, windows or dropDuplicates. That is where salting is still needed.

Salting a join and an aggregation

from pyspark.sql import functions as F
N = 16

# JOIN: big side gets a random salt; small side is replicated once per salt
rides_s = rides.withColumn('salt', (F.rand() * N).cast('int'))
drivers_s = drivers.crossJoin(spark.range(N).withColumnRenamed('id', 'salt'))
joined = rides_s.join(drivers_s, ['driver_id', 'salt']).drop('salt')

# AGGREGATION that partial-agg cannot shrink (e.g. distinct riders per driver)
step1 = (rides.withColumn('salt', (F.rand() * N).cast('int'))
              .groupBy('driver_id', 'salt')
              .agg(F.collect_set('rider_id').alias('riders')))
step2 = (step1.groupBy('driver_id')
              .agg(F.array_distinct(F.flatten(F.collect_list('riders'))).alias('riders')))

Salt only the hot keys in practice — when(col('driver_id').isin(hot), rand_salt).otherwise(0) — and replicate only those keys on the small side. Salting everything multiplies the small side by N for no reason.

At 100×

Skew grows faster than data. A key at 2% of a 10GB table is 200MB; at 2% of 10TB it is 200GB in one task, which spills or dies. Jobs that were fine for a year fall over without any code change.

Pick N from sizes, not vibes. Hot key = 200GB, target 200MB per task → N ≈ 1,000 for that key only. The small side for that key grows ×1,000 — fine for one driver's row, ruinous for the whole drivers table.

Measure keys up front. A daily groupBy(key).count().orderBy(desc('count')).limit(20) on the big join keys catches new hot keys before production does.

Watch AQE's thresholds on big clusters: with a 64MB advisory size and 256MB threshold, a 'mere' 2GB partition among 50MB ones is split; a 10GB one among 3GB ones (< 5× median) is not.

Gotchas

• Scaling the cluster — idle executors cannot share one task
• Raising shuffle partitions to fix skew — the hot key still hashes to exactly one partition, now among more tiny ones
• Assuming AQE fixes skewed groupBy — it handles joins; aggregations rely on partial aggregation, which fails for collect_list, countDistinct and windows
• Salting with rand() without salting the other side — the join loses matches; every key+salt on the big side needs its partner on the small side
• Salting every key — the small side multiplies by N; salt only the hot ones
• Forgetting nulls — in a left join, every NULL driver_id is hashed to the same partition even though none can ever match

A stage is as slow as its slowest task. Skew is one task doing everyone's work.

Filter junk keys → broadcast → AQE → salt. In that order.

Words to know

In two minutes

• Skew = one key holds a disproportionate share of rows; after hashing, one task gets it all and the stage waits for that straggler.
• Diagnose in the Spark UI: task duration and shuffle read size where max ≫ 75th percentile; more executors never help.
• Common culprits: nulls, defaults like 'UNASSIGNED' or 'unknown', one giant customer or city.
• AQE (spark.sql.adaptive.enabled, default true since 3.2) coalesces small shuffle partitions, switches joins to broadcast at runtime, and splits skewed partitions in sort-merge joins.
• A partition counts as skewed when it exceeds both 5× the median partition size and 256MB (skewedPartitionFactor, skewedPartitionThresholdInBytes).
• AQE's skew handling covers joins, not aggregations or windows. sum/count aggregations are usually fine thanks to partial aggregation; collect_list, countDistinct, windows and dropDuplicates are not.
• Salting: add a random salt 0..N-1 to the big side, replicate the small side N times, join on (key, salt); for aggregations, aggregate by (key, salt) then by key.
• Cheapest fix first: filter or separately handle nulls and junk keys, broadcast the small side, then AQE, then salting.

Test yourself

There are 8 questions, 3 puzzles and 5 recall cards for this lesson in the app — the recall cards come back on later days, just before you would forget.