What About Data Skew?

Concepts covered: Partitioning Strategies

Every shuffle in this lesson hashed a key into 200 partitions and assumed the keys spread out. Skew is what happens when they do not: 1 title_id for a live event carries 40% of a night's playback rows, so the partition that key hashes into holds 40% of the bytes, and the task that owns it runs 40% of the join alone on 1 core while 199 cores finish and wait. The interviewer's version is usually 'the stage has sat at 199 of 200 tasks for 5 hours, what is going on', and the answer needs 3 parts: how you confirm it, which of 3 fixes you reach for, and what each costs.

Confirm it from 2 places

The task summary from the slow-job section already said it: max duration 5 hours 33 minutes against a median of 25 seconds, and 1 task reading 96 GB of shuffle against a median of 0.6 GB. The second number matters more than the first, because a slow task with a normal read size points at a bad node or garbage collection rather than skew. Then find the key, because the fix depends on how many hot keys there are and whether one of them is NULL:

SELECT
  title_id,
  COUNT(*) AS rows_per_key
FROM playback_events
WHERE dt = DATE '2026-10-22'
GROUP BY title_id
ORDER BY rows_per_key DESC
LIMIT 10

Fix 1: adaptive skew join

Adaptive query execution is on by default since Spark 3.2, and its skew join handling has 2 conditions, both default: a shuffle partition counts as skewed when it is larger than 5 times the median partition and larger than 256 MB. Spark then splits the 96 GB partition into pieces near its advisory size and pairs each piece with a copy of the matching partition from the other side, so the join runs as many tasks instead of 1. It costs nothing to enable and works for sort-merge joins, which makes it the first answer. It also did not fire last night, and the reason is the shape of stage 3: the aggregate on title_id reuses the join's partitioning, so splitting the hot partition would force a second shuffle, and Spark skips the split rather than add one. Setting spark.sql.adaptive.forceOptimizeSkewedJoin accepts the extra shuffle. An aggregation on its own is safe: a sum or count runs as a partial aggregate before the exchange and a final one after it, so a hot key's rows collapse to 1 row per map task before they move. Salting covers the rest: joins adaptive execution will not split, aggregations whose partials do not shrink, and Spark versions before 3.0.

Fix 2: salt the key

from pyspark.sql import functions as F

SALTS = 64

salted_events = events.withColumn(
    'salt', (F.rand() * SALTS).cast('int')
)
salted_catalog = catalog.withColumn(
    'salt', F.explode(F.array([F.lit(i) for i in range(SALTS)]))
)
joined = salted_events.join(
    salted_catalog, ['title_id', 'salt']
).drop('salt')

The large side gets a random integer from 0 to 63 appended to its key, so the hot key hashes into 64 partitions of 1.5 GB instead of 1 of 96 GB, and 1.5 GB sits under the 2.2 GB task budget from the sizing section. The small side is exploded 64 times so that every salted key still finds its match, and that is the price: a 30 GB catalog becomes 1.9 TB of shuffle, which is why the catalog gets projected to the 3 columns the join needs first, at which point it weighs 2 GB and the explosion costs 128 GB. Pick the salt count from the UI: the hot partition's bytes divided by the per-task budget, rounded up to a power of 2. A plain sum or count on the hot key needs no salt. A distinct count over a huge group does, because its partials do not shrink: salt by a hash of the counted column, count distinct per (title_id, salt), then sum the 64 partials.

Fix 3: isolate the hot keys

With 1 or 2 hot keys known in advance, split the join. Rows whose title_id is in the hot set join the 2 matching catalog rows through a broadcast, so the large side never shuffles for them; every other row takes the sort-merge path, now with no partition above 2% of the total; union the 2 results. It costs 2 scans of the large side instead of 1, which pays off when the hot keys stay hot from day to day and loses when they change nightly, so it sits behind the adaptive fix. The NULL key is its own case. In an inner join it matches nothing, and Spark infers IsNotNull on the join key and drops those 10% before the shuffle, which is why no NULL partition shows in the task summary. In a left or outer join the rows must survive and all hash to 1 partition, so filter them out before the join and union them back with empty catalog columns.

About this lesson

This section is part of the The Spark Deep Dive: Intermediate lesson on DataDriven, the home of data engineering. Lessons cover SQL, Python, Spark, data modeling and pipeline architecture, and each section pairs its explanation with worked examples and practice you run as you go.