One Stage Dominates

Concepts covered: Spark UI Diagnosis

The commonest shape in this question is the one from the narration: 1 stage holds nearly all the job, and inside it 1 task holds nearly all the stage. The 199 ordinary tasks finish within the first few minutes and their executors sit idle for nearly 3 hours waiting for 1 task, because the stage cannot end until every task ends and the next stage cannot start until the stage ends. The cluster is 99% idle and the bill is 100% running. The shape has 1 cause in a join or an aggregate: a key value that too many rows share, so the task that owns that value's partition receives all of them.

The animation below turns off broadcasting, so the join runs as a sort-merge join and both sides shuffle on the key, and it gives the events table a key value that most rows share. Watch which task receives those rows and when the stage finishes.

spark.conf.set("spark.sql.autoBroadcastJoinThreshold", -1)
events.join(titles, "title_id", "left").count()

Seven of the 12 event rows carry no title_id. A left join keeps them, every one hashes to the same partition, and 1 task receives at least 7 of the 12 rows. Scale that ratio to a day of viewing events and the task that owns the empty key reads 12.4 GB while the median reads 135 MB. Without adaptive skew handling, the stage's duration is that task's duration however many executors the cluster has, because no other task can take a share of its partition. Adaptive execution, when enabled, can split a skewed join partition into several tasks, so 1 task of 2 h 48 min in the UI means it was off or did not apply. Splitting the empty rows out still matters: it removes their join work rather than spreading it.

Find the hot value

Before any fix, confirm which value is hot, because the fix depends on it. One aggregate over the join column answers the question, and it is the first query to run after the UI has pointed at the stage. Say the query in the interview. It is a checkable step rather than a hunch, and the interviewer can picture its result.

SELECT
  title_id,
  COUNT(*) AS event_rows
FROM viewing_events
WHERE event_date = DATE '2024-03-14'
GROUP BY
  title_id
ORDER BY
  event_rows DESC
LIMIT 5

The hot value is empty. A client release 3 days ago changed the app-open event: it used to carry the title the viewer last watched and now arrives with no title, so rows that always existed now share 1 key, and they make up 31% of the day. The input barely grew and the code did not change; what changed is how the rows spread across keys, which a byte count cannot see and the task summary can. These rows need no catalog lookup at all: a left join on an empty key matches nothing and returns the catalog columns as empty. So the fix removes them from the join, adds the empty columns directly, and unions the 2 halves back together.

with_title = events.filter(events.title_id.isNotNull())
no_title = events.filter(events.title_id.isNull())

joined = with_title.join(catalog, "title_id", "left")
padded = no_title.selectExpr(
    "*",
    "CAST(NULL AS STRING) AS title_name",
    "CAST(NULL AS STRING) AS genre"
)
result = joined.unionByName(padded)

A third of the stage's join work was for rows that needed none of it. After the change the join stage reads 26.8 GB instead of 39.2 GB, every task reads near the median, and the stage finishes when its slowest ordinary task does. The same input, re-run, gives the before-and-after that the method ends with, and the table is what you would describe in the room.

When the hot value is a real key rather than an empty one, a live event title that 30% of viewers watched, filtering is wrong because those rows need the catalog. Broadcasting the catalog avoids the shuffle, but 1.2 GB is over 100 times the 10 MB threshold: the driver would collect it, and every 8 GB executor would hold a full copy plus its hash table beside 4 task shares of roughly 1 GB, which risks killed executors. Salting fits better: give the hot key's events a random suffix from 0 to 9, copy that 1 catalog row 10 times with each suffix, set suffix 0 on every other row, and join on key and suffix, so 10 tasks share the hot rows. Broadcast only if projecting the catalog to the columns the job needs shrinks it enough. Then say the measurement: the max task within 2x of the median on the re-run.

About this lesson

This section is part of the Make It Faster 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.