Writing shuffle data: hash, sort, spill and consolidation

What decides how many shuffle files there are, and how big?

Hash shuffle versus sort shuffle

Both perform the transpose on the producer side. They differ in whether the transpose produces files or an index.

Hash shuffle

Hash shuffle does the work eagerly: for each row, compute the target and append the row to that target's stream. Each producer holds R open output buffers and writes R files. All its costs multiply with R, and that is what killed it:

File count = M × R. At 10,000 × 10,000 that is 10⁸ files. No file system's metadata path survives that.
Open file descriptors = R per concurrent task. Typical limits are 1,024 to 65,536, so R = 10,000 can fail immediately, before any performance question arises. This is Mateo's "too many open files".
Buffer memory = cores × R × buffer size. On a 16-core executor (a worker process that runs tasks on one machine) with R = 1,000 and 32 KB buffers: 16 × 1,000 × 32 KB = 512,000 KB ≈ 512 MB for shuffle buffers alone. This is the sharper constraint, because unlike file count it cannot be fixed with a better file system.

Sort shuffle

Sort shuffle reorders each task's output by target partition, so the transpose becomes an offset table: one data file per producer plus an index of R offsets. File count falls from M × R to M, open descriptors to one, and buffer memory to one buffer regardless of R. The consumer still issues M reads (one range per producer), so the read side is unchanged. Everything on the write side improved.

sort shuffle output of one producer:
  data file:   [partition 0 rows][partition 1 rows] ... [partition R-1 rows]
  index file:  offset(0), offset(1), ..., offset(R)      # R+1 numbers
consumer j reads bytes [offset(j), offset(j+1)) from each of the M data files

Two corrections to how this trade is usually described

Sort shuffle's "sort" is not a real sort. The sort key is a partition id, a small integer in [0, R). So the reorder is a counting pass (or radix pass): one sweep to count rows per partition, one sweep to place them. It costs O(n) with R counters and no comparisons. The statement "sort shuffle adds an O(n log n) sort" is simply wrong.
If the consumer needs sorted input (a sort-merge join, a sorted aggregation), sorting by key on the map side has a negative cost. The work was required anyway, and doing it here means it happens already partitioned, already parallel, and on smaller inputs. That is why sort-based shuffle and sort-based operators were developed together.

When hash still wins

Hash shuffle still wins when R is small. Below the descriptor and buffer limits, skipping the reorder is cheaper than doing it. That is why engines keep a bypass path that is used below a threshold in the low hundreds of partitions. It is a genuine crossover, not a legacy leftover.

The consequence promised in Skew and elasticity

Sort shuffle's output is contiguous and addressable by index for each partition. So a consumer can be assigned a byte range, or a sub-range of partition ids. That is the capability that adaptive skew splitting depends on. Hash shuffle's per-pair files can only be split at file granularity. A runtime skew fix was made possible by a file layout decision. This is the tightest coupling between two topics here, and an example of the principle "granularity is the floor on adaptivity": here the granularity is a property of the format.

Map-side spill and merge

The transpose needs somewhere to hold rows between "decide the partition" and "write the output". When a task's output exceeds its buffer B, the buffer must be drained. The drained content is already ordered by partition, so the natural drain is a sorted run (a sorted chunk written to disk). The producer therefore behaves like an external merge sort: fill B, order by partition, write a run, repeat, then merge all runs into the final indexed file.

The price is write amplification, and it can be computed

Number of runs = S_task / B (where S_task is the task's output size).
The merge fan-in k (how many runs one merge pass combines) is limited by read buffers: k × (read buffer size) ≤ B. With B = 64 MB and 1 MB read buffers, k ≤ 64.
Total passes = 1 + ⌈log_k(S_task / B)⌉. Each pass writes the data once.

This is how a memory shortfall turns into I/O cost at a known exchange rate. It is the principle "costs are moved, not removed" with the invoice attached. (We use external merge's arithmetic only to price the shuffle. The merge algorithm itself belongs to query-processing algorithms, and memory machinery such as grants, arenas and spill thresholds belongs to memory management.)

Partial aggregation before the spill

The other thing that happens on the map side is the only mechanism in this article that attacks S itself, not the denominators: partial aggregation (also called a combiner) before the spill. Each map task combines rows with the same key locally before writing them.

Its value is the local reduction ratio:

If the ratio is large, combining shrinks S in the same proportion, and every downstream cost shrinks with it. Example: a task has 1,000,000 rows but only 1,000 distinct meter_id values: ratio 1,000, so the output is 1,000 times smaller. If every row has a distinct key, the combiner buys nothing and costs a hash table. That is why engines monitor the ratio and abandon the combiner when it does not reduce. The criterion is simply rows_per_task / distinct_keys_per_task ≫ 1, and it can be checked on a sample. The enabling condition is algebraic: only associative reductions can be pre-combined at all.

Shuffle file consolidation

The M in M × R counts tasks, not machines. Tasks run one after another on the same cores. So several tasks on one executor can append to a shared file for each (executor, partition) pair. File count falls from M × R to E × R (where E is the number of executors), and the mean file size rises by M/E.

Two axes to collapse. There are exactly two axes along which the M × R product can be collapsed:

Sort shuffle merges along R: one file per task, indexed by partition.
Consolidation merges along M: one file per executor per partition, indexed by task.

Sort shuffle ends at M files and consolidation at roughly E × R, each plus an index. They differ in which index you keep, and so in which dimension remains splittable.

The cost is recovery bookkeeping. Concurrent appends to a shared file need handle pooling and mutual exclusion. More important, the one-file-per-task property that made retries trivial is gone. A failed task's partial appends are now mixed with valid data. So the index must record per-(task, partition) offsets, and readers must skip attempts that never committed. That index is the principle "immutable output grows an indirection layer plus a garbage collector". The offsets also show commit arbitration at work: a task's bytes count only once its index entry is registered. The registration is the single atomic act that speaks for everything the task wrote.

Historical note. Consolidation was a transitional mechanism. Sort shuffle made it redundant and it was removed from engines that had it. Its shape survived and is now the standard layout for external shuffle services. That is the better reason to remember it: not as a technique to use, but as the idea that the M axis can be collapsed.