What a Shuffle Moves Between Map and Reduce Stages

A map function can run record by record, on whatever split it was handed, and never needs to know that other tasks exist. A reduce function cannot: to compute a sum, a set union, or a join, it has to see every record that shares a key, and those records are scattered across every map task in the job. The shuffle is the machinery that gathers them, and it is the most expensive part of a batch job because it converts a record-parallel computation into a network- and disk-bound one.

Why processing has to stop at the boundary

The constraint is informational rather than mechanical. A reduce task cannot emit a correct result for key $k$ until the last map task that will ever produce a record for $k$ has finished. Until then, a sum might still grow and a join might still receive a match. There is no way to complete a reduce task early, and there is no way to start reduce tasks meaningfully before the map stage has ended, except to the extent that partial aggregation has already reduced what might arrive.

That property is what makes the stage boundary a barrier. Frameworks schedule the map stage, wait for its last task to report completion, then schedule the reduce stage. Each reduce task fetches the segments addressed to it from every completed map task. The stage split also means the map stage’s slowest task determines when the reduce stage begins, which is why skew in one partition delays everything downstream rather than only itself.

The map side: buffer, sort, spill, merge

A map task does not write a file per partition as it goes. Records go into a fixed-size in-memory buffer along with the partition number computed by the partitioner. When the buffer fills past its threshold, the task sorts the buffered records by partition number and then by key, and writes the result as one run to a local disk file. The buffer is then reused.

At task completion the runs are merged into a single output file per map task, laid out as $R$ consecutive segments, with an index recording the offset and length of each. The index is what allows a reduce task to request exactly the bytes addressed to it rather than the whole file.

Two properties of this file matter later. It is written to local disk, not to the distributed filesystem, so it is unreplicated and disappears with the machine — the reason a lost shuffle has to be recomputed. And it is written twice for every spill: once when the buffer overflows and again when the runs are merged into the final file. The number of spills does not change that total, but it does change how many runs the final merge has to combine, as the map-side buffer budget analysis shows.

The reduce side: fetch, merge, reduce

A reduce task for partition $p$ issues one fetch per map task, for the $p$-th segment of each. Fetches go over the network, usually over HTTP, with retries on failure, and a shuffle service can serve a segment after its map container has exited. Every segment is sorted internally, so the $M$ segments for a reduce task form $M$ sorted runs over disjoint key ranges.

The reduce task merges them by key. Small segments are merged in memory; when the accumulated data exceeds a configured share of available memory, the task performs a first pass that merges groups of runs to disk and a later pass that merges the results. The reference implementation combines on the order of ten runs per merge pass by default, so the number of merge passes grows logarithmically with the number of map tasks: a thousand map tasks need two or three on-disk passes rather than one, and each extra pass reads and writes the whole dataset again.

The merged, sorted stream is then fed to the reduce function one key group at a time, with the values for a key presented as an iterator. Because the group’s values arrive in sorted order, the reduce function can stream them without buffering, which is what keeps the reduce side’s memory independent of group size — unless the function itself collects the values, at which point the group must fit in memory.

What determines how much crosses the wire

Every record a map emits is written once to local disk and fetched exactly once by its reduce task, so the total shuffle volume equals the sum of the map outputs, and the reduce task count $R$ does not change it. A combiner lowers it by collapsing repeated keys before the write, which is the whole point of pre-aggregating map output.

What does not scale with volume is per-record and per-segment overhead. Shuffle records are serialized, typically with some bytes of length and type framing per record, and compressed when a codec is configured, so a codec can cut the bytes actually transferred by a large factor at the cost of map-side CPU. Segments impose a fixed cost per request: a job with 1,000 map tasks and 200 reduce tasks creates 200,000 segments and exactly that many fetch units, and if the average segment is small the per-request cost dominates the payload.

Compression sits on both sides of the barrier and interacts with both. Compressing map output cuts the bytes written to local disk, the bytes fetched over the network, and the bytes read back during the reduce-side merge, at the cost of compressing once and decompressing once per consumer. A codec that decompresses several hundred megabytes per second per core keeps that cost well below the I/O it removes; one that decompresses at tens of megabytes per second becomes the bottleneck itself. The two writes do not have to be treated alike, since spill runs are written and immediately read back by the same task, so a codec can be applied to the final map output while the intermediate runs stay uncompressed.

The ratio of shuffle bytes to input bytes depends entirely on the map function. A projection or filter emits fewer bytes than it consumes; a join of a small dimension against a billion rows emits about the same number of rows it read; an operation that expands a row into many rows, such as unnesting a nested array, can make the shuffle several times the input. In every case the number is fixed before the reduce stage sees it, which is why the map stage is where shuffle reduction has to happen.

Draining, retries, and the cases that break

Fetch failures are the most common visible symptom. A reducer that cannot read a segment retries, and after repeated failure reports the map task as failed rather than failing itself, at which point the map is re-executed and the reduce stage is restarted — a small machine loss can therefore cost an entire stage’s work. A federated detail worth knowing is that the coordinator tracks completion notifications per map attempt, so it knows which segments exist and where.

Several edge cases produce degenerate behaviour. A record larger than the sort buffer cannot be spilled normally and has to be written through; a partition receiving no records still incurs a fetch attempt against every map output, though frameworks skip empty segments when the index says the length is zero. Reduce functions that read their iterator twice see an empty second pass in implementations that stream from disk rather than from a materialized list. And a reduce task whose input is a single giant key group degenerates into a single-threaded loop over that group, because the framework parallelizes across keys, never within one.