Map-Side Buffer Budgets and the Cost of Spilling

A map task’s output does not go straight to the shuffle file. It accumulates in memory until a threshold is reached, and only then is it sorted and written out as a run. That buffer is the only place in the map stage where memory buys throughput, and its size is bounded by a number chosen before the job starts rather than by what the data would prefer.

A budget per task, not per worker

The buffer belongs to a task, and a worker usually runs several tasks at once. Eight concurrent map tasks with a sort buffer of a few hundred megabytes each would consume a worker’s entire memory before any user code ran, so frameworks express the buffer as a fraction of the container’s memory or as a fixed value in the tens to low hundreds of megabytes. A worker with four slots and a 256 MiB buffer devotes one gigabyte to buffers alone, and the cluster manager enforces the total container limit regardless of the buffer setting, so a misconfigured buffer does not fail gracefully — it fails when the operating system or the manager kills the container.

The buffer also competes with everything else in the same process: the input reader’s own read buffers, the deserialization path, any reduce-side fetch buffers in a combined task, and user code that collects data. Out-of-memory failures in shuffle-heavy jobs are more often caused by a reduce function building a list per key group than by the sort buffer, which is why enlarging the buffer does not reliably fix them.

Within the map task itself, a larger buffer is not unconditionally better. The spill is a pause: the task must sort the buffered records before it can write them, and while the sort runs, no new records are being produced. A buffer twice the size holds twice as many records but also stalls the map function for roughly twice as long when it finally flushes. The total sort work over the task’s life is the same either way, since it is proportional to the number of records times the logarithm of the number held at once, so the buffer size trades stall granularity and merge depth rather than total CPU.

What happens at the threshold

The spill is triggered at a configured percentage of the buffer rather than at its exact limit, commonly around eighty percent, so that records continue to be accepted while the buffered ones are being sorted and written. The task sorts the buffered records by partition and key, writes them sequentially as one run to local disk, and clears the buffer for reuse.

The I/O arithmetic here is frequently misstated. Each record enters exactly one spill run, so the total bytes written across all spills equal the total map output bytes. The final merge into the single map output file then writes those same bytes a second time. The map side therefore writes its output twice regardless of whether the task spills once or thirty times. What grows with the spill count is the number of runs the final merge must combine — and therefore the number of merge passes, since the merge combines on the order of ten runs at a time. A task with thirty spill runs needs two merge passes and rewrites its own output once more per pass, so the write volume becomes three times the map output instead of two.

Small spills have two further costs. Compression is applied per run or per file, and a codec that compresses a few hundred kilobytes of data reaches a worse ratio than one compressing tens of megabytes, because its dictionary and window have less context. And each run carries framing overhead — headers, an index entry, a file to open — which is fixed per run rather than per byte.

Read amplification on the reduce side

Every reduce task merges one segment from each map task, so the merge width equals the map task count. With a merge factor of about ten, a reduce task facing 1,000 segments needs three passes over its data, each pass reading and writing the whole partition. A job with 4,000 map tasks needs four. The reduce side, unlike the map side, cannot avoid this by buffering more: the segments are already on disk at a fixed width, and the number of runs is a property of the map stage’s parallelism rather than of the reducer’s memory.

The reducer does hold a portion of its input in memory before spilling its own merge output, and that buffering is where a reduce-side memory budget is spent. The relevant quantity is the fraction of reduce memory reserved for in-memory merging; when a partition’s data exceeds it, the merge goes to disk and each subsequent pass costs a read and a write of the same records. Compressing both map output and reduce-side spills cuts those bytes at the cost of CPU on both sides, which is why a codec that is expensive per byte can still reduce wall-clock time when the bottleneck is disk or network rather than processor.

The reduce stage’s own output has the same granularity problem one level further down. Each reduce task writes one file per partition of the stage’s output, so a job with a thousand reduce tasks emits a thousand files for a downstream reader to split, and no amount of spill tuning changes that count. The buffer is a throughput knob, not a layout knob: it determines how many runs each task merges and how much extra I/O the merge performs, while the file count is fixed by the partition count chosen in the partitioner’s arithmetic. Mixing the two decisions tends to produce a job with an oversized buffer and an unchanged file layout.

Estimating the spill count before choosing a size

The spill behaviour of a task is predictable from two numbers: how many bytes the map will emit, and how many bytes the buffer holds. A map producing 4 GiB of output with a 256 MiB buffer spilling at eighty percent accumulates about 20 runs. With a ten-way merge factor that is two passes, and the difference between a 128 MiB and a 512 MiB buffer for the same task is one merge pass at most.

The estimate breaks down on record size. An in-memory record is not the same size as its serialized form: object headers, references, and per-field padding commonly make the in-memory representation several times larger than the bytes that reach disk, so a buffer measured in bytes holds far fewer records than its size divided by the serialized record length suggests. Frameworks that store intermediate records in an encoded form inside the buffer, and compare encoded keys directly, fit several times as many records in the same budget.

Two edge cases distort the estimate further. A single record larger than the buffer cannot be accumulated and must be written through directly, so a task processing a few very large values spills constantly no matter how the buffer is sized. And a task whose output is small but whose input is large — a filter that discards most records — benefits from a large input reader budget and no spill capacity at all, which is the opposite of the default allocation. The same reasoning about per-task budgets applies on the input side described in input splits and task granularity, where the unit sizes are chosen before the job runs, and the cost of getting them wrong is paid in the shuffle data path on every record that crosses the barrier.