What makes a Spark shuffle task spill records to disk?

## Spill is a safety valve, not an error A Spark task processing a shuffle cannot assume its data fits in memory. Rather than failing, the memory-hungry structures — sorters, aggregation maps, join buffers — are *spillable*: when they cannot get more memory, they serialize their current contents to local disk in sorted order and start fresh. Spill costs a write and a re-read, plus a merge, so a spilling stage is slower, but it completes. A stage that spills a little is usually fine; a stage spilling tens of gigabytes is telling you a sizing decision is wrong. ## Where spill happens in a shuffle **Write side.** The shuffle writer inserts records into an `ExternalSorter` (or `ShuffleExternalSorter` on the serialized path), which sorts by partition id — and by key when the operation does map-side combine. Whenever it fails to acquire additional execution memory, it sorts the buffer, writes that run to a spill file under `spark.local.dir`, and keeps going. At task end, all runs merge into the map task's single data file. `spark.shuffle.spill.compress` (on by default) compresses those runs. **Read side.** Reduce tasks spill too. An aggregation whose per-key state does not fit uses a spilling append-only map; a sort-merge join sorts each side with a spilling sorter. This is why you can see spill in a stage that performs no shuffle write at all. ## Why free cluster memory does not prevent it The memory that matters is a **task's share of one executor's execution pool**. Spark's unified memory manager splits the executor heap into an execution region (shuffles, sorts, joins, aggregations) and a storage region (cached blocks); execution can borrow back from storage by evicting cached blocks, but it can never reach across to another executor. Within an executor, the pool is divided among the tasks running concurrently, which is `spark.executor.cores` of them. So an executor with a large heap and 8 cores gives each task roughly an eighth of the pool, and a job can spill heavily while the cluster-wide memory chart looks half empty. The corollary surprises people: *reducing* cores per executor often removes spill, because each remaining task gets a bigger slice. ## Reading the metrics The stage detail page reports two numbers per task, and they are not measuring the same thing: - **Spill (Memory)** — the estimated size of the records *as held in memory* before spilling: deserialized objects, with JVM overhead. - **Spill (Disk)** — the bytes actually written after serialization and compression. Disk is normally several times smaller than memory, and that ratio is a rough gauge of how expensive your in-memory representation is. Compare spill against shuffle read size for the same stage: spilling twice the data you read means it is being written and re-read repeatedly through multiple merge passes. ## Fixing it, in order of leverage 1. **More shuffle partitions.** Halving the bytes per partition halves what each task must hold. This is the first lever because it is cheap and reversible, and it is why an under-set partition count shows up as spill first. 2. **Fewer cores per executor** (or more memory per executor). Directly enlarges each task's slice of the execution pool. 3. **Pre-aggregate.** `reduceByKey` and `aggregateByKey` combine on the map side; `groupByKey` moves every raw record and materialises whole groups on the reader, which is the textbook way to force enormous spill. In DataFrame land the planner already does partial aggregation, but a wide `collect_list` reproduces the same problem. 4. **Shrink the rows.** Project away unused columns before the exchange; fewer bytes cross and fewer bytes must be held. 5. **Address skew** if only a handful of tasks spill while the rest do not — that is a distribution problem, not a sizing problem. Raising `spark.memory.fraction` is available but low-leverage and risks starving cached data; treat it as a last resort rather than a first answer. ## What interviewers are checking The distinguishing insight is *per-task* memory. A candidate who says "add executors" has not understood that spill is decided inside one task's memory budget, and more executors change neither the bytes per partition nor the slice per task unless the partition count moves with them. The second insight is that spill is a graceful degradation — the alternative would be an OOM, and tuning that eliminates all spill at the cost of huge executors is often the worse trade.