Shuffle Fault Tolerance: Recomputing Lost Map Output

Input data in a distributed filesystem is replicated and survives the loss of a machine. The shuffle output produced from that input does not. It is written to the local disk of whichever worker ran the map task, unreplicated by design, and read directly from there by every reduce task that needs it. Recovery therefore has to reconstruct something the system deliberately declined to store, and the cost of that reconstruction is the subject of this article.

Map output is a derived cache on local disk

When a map task completes, the coordinator receives a notification that identifies the attempt and the location of its output file on that worker’s local disk. Reduce tasks use those locations directly, fetching each partition’s segment over the network, usually by an HTTP request handled by the worker’s process. Nothing about this path involves replication.

Replicating shuffle output is not obviously wrong, but it is rarely worth it. The output is derived: the input split it came from remains readable, so it can always be recomputed. Replication would triple the disk footprint and send the bytes over the network twice, once for the write and once for the read, in exchange for avoiding a recomputation whose cost is the map’s own runtime. The trade only inverts when map tasks are long — a map that reads a multi-gigabyte split and applies an expensive transformation is genuinely expensive to rebuild, and that is the case where persisting shuffle data to a shared store starts to pay.

The practical consequence is that a machine loss is not merely a loss of that machine’s compute. It invalidates every shuffle segment the machine hosted, including segments from map tasks that completed long ago and whose results were already consumed in part.

Recomputing from the input

The default recovery is re-execution. A reduce task that cannot fetch a segment after retrying does not fail; it reports the map attempt as failed to the coordinator, which marks the map output invalid and re-runs the map task, normally on a different worker. The old output is then useless, and every reduce task that had already fetched and processed data from it must be re-executed, because its partial results are based on input that the recomputed map may produce differently.

That second part is what makes the cost disproportionate. In the reference MapReduce design, a fetch failure can invalidate an entire reduce stage: not only the map is re-run, but all reduce tasks that touched its output, and the reduce stage re-executes from the beginning. Losing one machine near the end of a large job can therefore discard the whole reduce stage’s progress, and the failure is attributed to the map attempt rather than to the reducer that detected it, which is why fetch failures in logs appear as map failures.

Recomputation also assumes the map function is a pure function of its input. A map that writes to an external system, or that reads a value that changes between attempts, produces a different result the second time, and any downstream re-execution is then based on data the first run never had.

Shuffle services and materialization boundaries

An external shuffle service removes the tight coupling between map output and task lifetime. The service runs on each worker as a long-lived process, owns the shuffle files, and answers fetch requests independently of the containers that wrote them. Containers can then exit — because a task failed, because a speculative duplicate finished, or because the cluster manager reclaimed memory — without taking their shuffle output with them. The output survives every failure that does not kill the node itself, so it removes a large class of recomputation without adding network traffic.

What it does not survive is node loss, since the files are still local disk on one machine. Reducing that further means writing map output to a replicated store, which is the trade described above: durable, but with the shuffle bytes crossing the network on the write path as well as the read path.

A coarser answer is to make the shuffle itself the durability point. In a lineage-based system, each partition of a dataset records the operations that produced it and their dependencies. If a partition is lost, the engine recomputes that partition alone by replaying its lineage from the nearest available ancestor. Narrow dependencies, where each output partition derives from exactly one input partition, allow a single lost partition to be rebuilt without touching the others. Wide dependencies, which are exactly the shuffle boundaries, do not: if an output partition after a shuffle is lost, all parent partitions must be re-read to reconstruct it, so the entire shuffle is re-executed. That asymmetry is why persisting intermediate results across shuffle boundaries — checkpointing a stage to reliable storage, rather than holding it in memory — reduces recovery cost by truncating the lineage graph, and why a shuffle file, once written, behaves like a checkpoint even though it was never intended as one.

Why non-determinism breaks recovery

Recomputation assumes the second run produces the same partitions as the first. A hash function that changes between attempts, a partitioner that depends on ambient state, or a map that samples without a fixed seed can place a key elsewhere on the retry. The recomputed segment then does not contain what the original did, and a reduce task that re-fetches it may aggregate an incomplete key group while reporting success. Determinism of the partitioner is thus a correctness requirement rather than a performance property, and it is the reason partitioner choice is not a free tuning knob.

Failure modes that produce loops and duplicates

Repeated fetch failures from a node that is alive but slow create a loop: the reducer times out, the map is re-executed elsewhere, another fetch fails, and the cycle consumes slots without progress. Schedulers counter this by tracking repeated attempt failures per node and declining to place further attempts there, which isolates the slow machine at the price of shrinking the cluster it is scheduled on.

Speculative execution creates a different hazard. A straggling map may have two concurrent attempts, and when the first completes, the second is cancelled and its output discarded, so the coordinator must ensure reduce tasks fetch only from the attempt it has accepted as canonical. If a cancelled attempt has already announced completion and a reducer has fetched from it, the fetched data is a legitimate result of the same deterministic function, so the failure is one of bookkeeping rather than of correctness — but only if the function really is deterministic.

Output visibility is guarded separately. A re-executed reduce task that writes its results directly to the output location would leave duplicate or partial output visible to downstream readers, so frameworks write task output to a private staging area and move it into the final location at job commit, typically by an atomic directory rename. The shuffle path itself gives at-least-once delivery of records to the reduce function; exactly-once effects inside a task are a property of that commit protocol, not of the shuffle, and they do not extend to side effects a task performs outside the framework.

The granularity of recomputation is decided before the job starts, by the input splits. With a thousand small splits, losing a worker costs one map task’s work. With a single unsplittable input file compressed as one member, one map task holds the entire input and its failure means re-reading and reprocessing all of it — the same property that co-partitioned layouts exploit for reuse makes recovery coarse when the unit of work is coarse.