The Architecture Recital

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.

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.

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.

Walk the estimate the way the planner does. With the cost-based optimizer off, which is the default (spark.sql.cbo.enabled is false), a filter's output is estimated at the size of its input. The planner sees 2GB of events joined to 400MB of users, neither below the 10MB broadcast threshold, and picks a sort-merge join. That plan shuffles both sides by user_id and then shuffles again by country for the aggregation: 3 exchanges and 4 stages.

With column statistics and no histogram, the cost model estimates an equality filter as 1 divided by the column's distinct count. Events has 25 event types, so the estimate is 4% of 2GB, roughly 80MB. That shrinks the error from about 250 times to 10 times, and 80MB is still 8 times over the 10MB broadcast threshold. Purchases are 0.4% of events because event types are far from uniform: page views make up most rows and purchases are rare. An estimate that assumes every value is equally common has no way to see that. Histograms (spark.sql.statistics.histogram.enabled, off by default) would, at the price of a heavier ANALYZE that must be rerun as the data shifts.

Adaptive query execution, on by default since Spark 3.2, re-plans at each shuffle boundary from the sizes the finished map tasks report. Here, once the events map stage completes, AQE sees around 8MB of shuffle output, under spark.sql.adaptive.autoBroadcastJoinThreshold (which takes the 10MB static value when unset), and rewrites the sort-merge join into a broadcast hash join. Notice what it can't take back. The users map stage was submitted alongside the events stage and has usually already written its 400MB of shuffle files. AQE saves the sort and reads those files through a local shuffle reader instead of fetching them across the network. The exchange by country stays.

The 0.4% share and the 25 event types are this lesson's example. On your own tables, explain(mode="cost") prints the planner's estimates for the same comparison.

Statistics are a snapshot from the last ANALYZE TABLE. A dimension that has grown from 6MB to 900MB since then still reads as 6MB, so the planner broadcasts it. Spark falls back to file sizes for a table that was never analyzed, but once a size is stored the planner uses it, and automatic size updates (spark.sql.statistics.size.autoUpdate.enabled) are off by default. The direction of the error decides the damage: an overestimate costs a shuffle you didn't need, while an underestimate can broadcast a table that does not fit in memory.

For country = 'JP' AND language = 'ja', the cost model multiplies the 2 selectivities. If each matches 3% of 50 million users, the estimate is 0.09%, or 45,000 rows. The columns describe nearly the same fact, so the real answer sits close to 3%, about 1.5 million rows: a 33 times underestimate, in the direction that ends in a memory error.

The skew rule fires when a partition passes 2 tests at once: larger than 5 times the median partition (spark.sql.adaptive.skewJoin.skewedPartitionFactor) and larger than 256MB (spark.sql.adaptive.skewJoin.skewedPartitionThresholdInBytes). Take a join whose median reduce partition is 200MB and whose hot partition is 12GB. The rule cuts the hot side into pieces near the 64MB advisory size, up to about 192 of them, bounded by how many map tasks wrote into it. Each piece is joined against the whole matching partition from the other side, so a 300MB partition there is read around 192 times: roughly 57GB of shuffle reads to fix a 12GB hot spot. The trade usually wins because those reads run in parallel across the cluster, and it explains why a join stage's read total can dwarf what was written to it.

One added sentence repairs all 3 failing checks: 'The planner assumed the purchase filter kept all 2GB of events, the job measures about 8MB, and I would confirm it by comparing explain(mode="cost") with the output rows on the SQL tab.' The tour keeps its order and its length stays under a minute; it gains the evidence an interviewer would otherwise spend several questions extracting.

A tour with no failure modes still fits a 60-second opener. The miss is offering it as the whole answer when the question asked how the system behaves on a particular query.

A nightly job joins a day of order events to customers, groups by order_id, and runs a Python UDF that parses each order's JSON payload into 40 fields. The team had tuned spark.sql.shuffle.partitions to 400 for this job on a cluster with 200 executor cores, so the UDF stage ran 400 tasks in 2 waves of about 2.5 minutes each, roughly 5 minutes for the stage. Then the platform team moved the cluster to a Spark version with adaptive execution on by default, and set spark.sql.adaptive.coalescePartitions.parallelismFirst to false to cut the count of tiny tasks on the shared cluster. The next night the job missed its deadline.

On the SQL tab, the exchange before the UDF had an AQEShuffleRead node above it reporting the coalesced partition count and roughly 1.2GB of shuffle data. The page showed the plan as AdaptiveSparkPlan isFinalPlan=true, and the initial plan beside it still read hashpartitioning(order_id, 400). The event timeline showed most executors empty for the length of the stage, and the task summary showed durations clustered tightly: no long tail, every task uniformly slow.

With parallelismFirst off, coalescing packs adjacent shuffle partitions together until each reaches spark.sql.adaptive.advisoryPartitionSizeInBytes, 64MB by default. The job wrote roughly 1.2GB of shuffle output, and dividing that by 64MB gives roughly 19 partitions. The tuned 400 had become only the starting count. The flaw is in what the 64MB measures: compressed shuffle bytes, the input to the stage. The expensive work in this stage was the UDF, whose cost follows row count and payload length, and repetitive JSON compresses very well. Each remaining task now carried about 21 times the rows of an old one, so a 2.5 minute task became a 53 minute task, while 181 of the 200 cores waited.

The team chose the repartition(400). It confined the fix to the single expensive step and kept coalescing for the cheap shuffles elsewhere in the job. The extra stage wrote less than the 1.2GB the aggregation had shuffled, a small price next to a 48 minute regression. They rejected turning coalescing off, because that fixes this stage and silently regresses every small shuffle in the job.