What does spill mean in Spark, and why is Spill (memory) bigger than Spill (disk)?

Spill is Spark's way of not failing. A task that sorts rows for a join, or groups them for an aggregation, holds them in execution memory. When they stop fitting, Spark sorts what it has, writes it to a file on the executor's local disk, frees the memory and carries on. At the end it merges the files back. The task finishes, but it has written and read data it would have kept in memory, and that costs time.

The 2 spill numbers

The Spark UI shows 2 figures for every stage and task that spilled, and they never match:

• Spill (memory) is how big the spilled rows were as objects in memory.
• Spill (disk) is how big the same rows were once turned into bytes and compressed on disk. spark.shuffle.spill.compress is on by default.

So the disk figure is the smaller one, often a third or a quarter of the memory figure. Both describe the same data. In the job we run on our Spark UI guide, a sort of 12 million orders over 4 partitions spilled about 228 MiB from memory per task, which landed as about 73 MiB on disk.

Where to see it

• Stage page. The summary metrics show Spill (memory) and Spill (disk) per task, from min to max. Every task spilling a similar amount means the partitions are too big. One task spilling while the median spills nothing means skewShuffle and Data SkewA shuffle moves rows between machines so rows with the same key meet, for a join or group by; skew is when one key has far more rows, so one task does most of the work..
• SQL / DataFrame tab. The Sort and HashAggregate boxes of the plan show spill size and peak memory, so you can see which step of the query spilled.
• Executors tab. Task time in red means garbage collection took over 10% of it, a common companion of spill.

Why it happens

Each executor shares its execution memory among the tasks running at once. With spark.memory.fraction at 0.6, an executor with 8 GiB gives about 4.6 GiB to execution and cached data together, and 4 cores split that 4 ways. A task spills when its partition is bigger than its share. The usual reasons:

1. Too few shuffle partitions for the data, so each partition is large. Our 4-partition sort spilled; the same sort over 64 partitions spilled nothing and ran in 625 ms instead of 1,132 ms.
2. One heavy key, which puts far more rows in one partition. In our skewed join one task spilled 148 MiB while the median task spilled none.
3. Wide rows, such as long strings or arrays, that make each row big.

What to do

Make each task's input smaller before giving it more memory, as Spark's own tuning guide advises. Raise spark.sql.shuffle.partitions, or let adaptive query executionAdaptive Query ExecutionSpark's way of re-planning a query while it runs: after each shuffle it looks at the real sizes, then merges small partitions, splits skewed ones and can switch a join to a broadcast. split a skewed partition. Select only the columns the join needs. Only then size up executor memory or reduce cores per executor (executor memorySpark Executor MemoryThe memory each Spark worker process gets: a Java heap, split into a shared pool for shuffles, joins and cached data and a pool for your own objects, plus overhead outside the heap. explains the split). A little spill is harmless; spill on every task of a long stage is time worth getting back. The shuffle partition calculator shows how big each partition will be before you run the job.