Where the Model Breaks
At this depth the interviewer takes for granted that you can trace a query from the driver to the executors. What they want to know is whether you understand that the trace describes decisions made once, on the driver, before any input has been read. Every choice in the physical plan, from the join strategy to the reducer count to which side becomes the hash table, is a bet placed on estimates. A recital scores at the points where those bets lose.
What the planner does read
The planner starts from real numbers about stored data. For a file-based table that has never been analyzed, it takes the total file size from the directory listing and uses it as the table's sizeInBytes. After ANALYZE TABLE, it reads the stored size and, with column statistics, the distinct count, min, max and null count of each column. Both are facts about the table at rest. Neither says how many rows survive a predicate in this particular query, how the join key is spread, or how large an intermediate result will be once it has been joined, filtered and aggregated.
3 facts the planner cannot know
| Runtime fact | What the static plan assumes | What actually happens |
|---|---|---|
| Selectivity of a filter | Without column statistics, a filter keeps its whole input size | `event_type = 'purchase'` keeps 0.4% of a 2GB table, about 8MB |
| Distribution of the join key | Hash partitioning spreads rows evenly across reducers | A bot account owns 30% of rows, so a single reducer gets 30% of the work |
| Size of each shuffle's output | `spark.sql.shuffle.partitions`, 200 by default, suits every shuffle | 200 reducers over 50MB gives 250KB tasks; over 2TB it gives 10GB tasks |
Each wrong assumption has its own price. A missed selectivity costs a shuffle that a broadcast would have avoided. A skewed key costs a straggler that holds the whole stage open while every other core idles. A wrong partition count costs scheduler overhead from thousands of tiny tasks in one direction, or spill to disk from a few huge ones in the other.
Watching skew land on a single task
The program below joins 12 orders to 5 users with automatic broadcast turned off, so Spark plans a sort-merge join and shuffles both sides by user_id into 3 reducer partitions. Adaptive execution is turned off as well (spark.sql.adaptive.enabled set to false), because on data this small its coalescing would merge all 3 reducers into 1 and hide the static plan. User 1 owns 8 of the 12 orders. Follow those 8 rows through the shuffle.
spark.conf.set("spark.sql.adaptive.enabled", "false")
spark.conf.set("spark.sql.autoBroadcastJoinThreshold", -1)
result = orders.join(users, "user_id").collect()| order_id | user_id | amount |
|---|---|---|
| 101 | 1 | 20 |
| 102 | 1 | 35 |
| 103 | 2 | 12 |
| 104 | 1 | 8 |
| 105 | 3 | 50 |
| 106 | 1 | 14 |
| 107 | 4 | 22 |
| 108 | 1 | 9 |
| 109 | 1 | 41 |
| 110 | 5 | 17 |
| 111 | 1 | 26 |
| 112 | 1 | 30 |
| user_id | country |
|---|---|
| 1 | US |
| 2 | JP |
| 3 | DE |
| 4 | BR |
| 5 | IN |
Every row for user 1 hashes to the same reducer, whichever reducer that turns out to be, because hash partitioning is a pure function of the key. That task handles at least 8 order rows while the other 2 share 4 at most. Nothing in the static plan reacts: the partitioner was fixed at plan time, and it has no way to split a key whose size it never measured. Scale the same shape to 30% of 4 billion rows over 200 reducers and the hot partition holds 1.2 billion rows against about 14 million in each of the others. With adaptive execution on, this join is the case its skew handling is built for: it cuts the oversized partition into several tasks and gives each one the matching users partition. Put the same hot key under groupBy("user_id") with collect_list, which partial aggregation can't shrink, and no rule splits it: one task receives 1.2 billion rows while the other 199 finish in seconds.
Where to look when you suspect the model broke
4 checks, in the order to run them
- Stages tab, task summary: compare the max task duration with the median. A max 20 times the median points at skew, and adding executors will not move it.
- Run df.explain(mode="cost") to print the size and row estimates the planner used, then compare them with each operator's number of output rows on the SQL tab.
- SQL tab, exchange nodes: shuffle bytes written total shows what each shuffle actually moved, the figure the partition count should have been sized for.
- Stages tab, spill columns: a nonzero Spill (Disk) on a reduce stage means its partitions outgrew execution memory.
Spark has 3 layers aimed at this gap, each acting at a different moment: rewrite rules that need no data, a cost model that reads stored statistics, and adaptive execution that re-plans between stages using measured output. Each closes part of the gap and leaves a blind spot of its own, and those blind spots are what the rest of this lesson takes apart.
About this lesson
This section is part of the The Architecture Recital 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.