Sort-Based and Hash-Based Shuffle Implementations

The map side of a shuffle has one job: get each record onto disk under the partition it belongs to, cheaply, so that the reduce side can find it. There are two standard ways to do that, and they differ in a single decision — whether the records are sorted before they are written. Everything else, including how many files exist at the end and how much work the reducer does, follows from that decision.

The file-count problem in hash-based shuffle

A hash-based writer divides the map’s output buffer into $R$ regions, one per reduce partition, and appends each record to the region for its partition without regard to key order. When the buffer fills, the writer flushes each non-empty region to its own file on local disk and reuses the buffer. At the end of the map task, the per-partition spill files are concatenated into one file per partition, or all partitions are merged into a single file laid out as $R$ segments.

The advantage is that no comparison happens. The map function’s output goes straight into a buffer slot and out to disk in the order it was produced. The disadvantage is the number of files. A map task with a buffer of size $B$ producing $S$ bytes of output spills roughly $S/B$ times, and each spill touches up to $R$ files, so the pre-merge file count is bounded by $R \times S/B$ and the post-merge count is $R$ per map task. For a job with 1,000 map tasks and 200 reduce partitions, that is 200,000 final segments before any spill multiplication, each one a separately opened and separately fetched file. Raising the partition count to 20,000, as jobs with tens of terabytes of input do, turns the same job into 20 million segments.

The consequences are not only metadata. A reducer fetching from many map tasks opens many files, and each fetch is a separate request with its own latency and connection cost. On spinning disks the reads are scattered across the device, so throughput drops to the seek-limited rate of a few hundred operations per second rather than the sequential rate. The coordinator also tracks a completion record per map output, and its bookkeeping and heartbeat traffic grow with the same product.

Sort-based shuffle: one file with an index

A sort-based writer ignores partition regions. Records go into a single buffer with their partition number attached; when the buffer fills, the writer sorts the whole buffer by partition number first and key second, then writes one run. At completion, the runs are merged in partition order into a single output file per map task, with an index giving the offset and length of each of the $R$ segments.

The file count becomes one per map task, independent of $R$. The index means a fetch for partition $p$ reads exactly one contiguous region, so the reducer’s read pattern is sequential and the per-request overhead is amortized over a segment rather than a file. Because each segment is sorted by key, the $M$ segments a reducer fetches for its partition arrive as sorted runs, and the reducer can merge them with a $k$-way merge instead of sorting them from scratch — a merge over $M$ sorted runs costs about $n \log M$ comparisons for $n$ records, against $n \log n$ for a full sort of the same records.

The price is the sort itself, which is performed on buffer-sized chunks rather than on the whole map output, so its cost per record is bounded by the log of the number of records a buffer holds. With a buffer holding on the order of a million records, that is roughly twenty comparisons per record, most of them cheap comparisons of a partition number that rejects the comparison in a single step. Modern implementations also sort serialized records in place and compare their encoded key bytes directly, which avoids materializing an object per record.

Spill merging and the arithmetic of runs

Both designs perform a final merge, but the sort-based design’s merge is a partition-ordered merge of already-sorted runs, so the cost is a linear pass over the spilled bytes with a heap of open runs. The number of simultaneously open runs is bounded by the merge factor, on the order of ten in the reference implementation, so a map task with thirty spills merges in three passes instead of one, reading and rewriting its own output each time. Merging during a spill, rather than only at the end, keeps the number of runs from growing linearly with the number of spills.

The run count interacts with the buffer size in a way worth stating explicitly: doubling the buffer halves the spill count and therefore removes one merge pass, but it also doubles the memory the task holds, and with several tasks per worker that is the axis on which the map-side buffer budget is set.

Where each design wins

Hash-based shuffle is competitive when the partition count is small. With a handful of reduce tasks there are only a handful of files per map task, the merge step is trivial, and the saved sort is pure gain. It is also simpler to implement when the map output is consumed by an operation that does not need order, such as hash aggregation on the reducer.

Sort-based shuffle wins as the partition count grows, because its file count and index size are independent of $R$. It also wins when the reduce operation can exploit sorted input: a merge join, a group-by that wants ordered output, or a reducer that merges runs rather than building a hash table. The design also composes with compression, because a codec can be applied to whole segment boundaries, and with spill compression, at the cost of CPU on the map side.

The hardware matters as much as the algorithm. Hash-based output with many small files is cheap on solid-state storage, where random reads cost about the same as sequential ones, and expensive on spinning disks, where the gap is two orders of magnitude. Sort-based output is sequential on both, so it is the design that scales across storage types and partition counts.

Real deployments moved from hash-based to sort-based as jobs grew to many reduce tasks, and the reverse move would require a good reason, principally a shuffle whose reduce side does no key-ordered work and a partition count small enough that the file count never becomes the bottleneck.

Contracts both implementations must honour

Neither design is free to differ in observable results. Both must guarantee that records with equal keys reach the same partition — the invariant behind the partitioner — and both must assign a key identically on a retried map attempt, or a reducer sees a partial key group.

The order of values within one key group is a different matter. A hash-based shuffle presents them in arrival order; a sort-based shuffle presents them ordered by key, with values in an implementation-defined order among themselves. Reduce functions must therefore treat the value sequence as unordered, and only operations that explicitly sort the value list are entitled to rely on ordering.

Empty segments need handling in both. A partition with no records still has an entry, either as a zero-length file or as an index record with length zero, and the fetch path should treat it as satisfied without a network request; otherwise a job with 1,000 map tasks and a thousand near-empty partitions spends its reduce stage issuing requests for nothing.

Segments also define the unit of compression and of retry. A codec applied per segment means a failed fetch retries one segment; a codec applied per file means a retry can pull in several partitions’ bytes. Neither is wrong, but the choice determines how much overhead a single fetch failure implies, which matters for the recovery cost after a lost map output.