The whole problem is one sentence: the data does not fit in memory. You have 1TB to sort and 16GB of RAM per machine, so the data is roughly 64x larger than the memory you get to sort it in. Every classic in-memory sort - quicksort, the sort() in your standard library - assumes random access to the entire array. The moment the array is bigger than RAM, that assumption is gone, and the algorithm either crashes with an out-of-memory error or thrashes the page cache into oblivion. So the real question is not “which comparison sort,” it is “how do you order data you can only ever see a small window of at a time.” The answer, since the 1960s, is external merge sort: read data in chunks small enough to sort in memory, write each sorted chunk to disk, then merge the sorted chunks back together with a streaming k-way merge that only ever holds one small buffer per chunk in RAM. Scale that across a cluster and you get the distributed sort that wins the TeraSort benchmark. This post builds both, single machine first because the distributed version is just external sort with the partitioning done up front.

The reason this is a good interview problem is that it forces you to reason about the memory hierarchy explicitly - RAM is fast and tiny, disk is slow and huge, network sits in between - and to design an algorithm whose cost is dominated by sequential I/O, not by comparisons. A naive “load it and call sort” answer dies instantly. Let me build it properly, end to end.

Functional Requirements (FR)

In scope:

  • Sort a 1TB dataset into one globally sorted output. Input is a 1TB file (or a set of files) of records; output is the same records, in sorted order by a specified key, written back to durable storage. Global order means if you concatenate the output files in order, the whole thing is sorted - not just sorted within each file.
  • Records with a sort key. Each record is a key plus a payload (say a 10-byte key and a 90-byte value, the classic TeraSort record shape, ~100 bytes total). Sort is by key, ascending, with a defined total order (lexicographic on bytes, or numeric). Ties broken by a stable rule if stability is requested.
  • Commodity hardware constraint. Machines have 16GB RAM each and ordinary local disks (HDD or SSD). No single machine can hold the dataset in memory. The design must work within that memory budget, on either one machine (external sort) or a cluster of them (distributed sort).
  • Correct and complete. Every input record appears exactly once in the output, in the right place. No records lost, none duplicated, none reordered incorrectly. This is a batch job with a definite end, verified.
  • Restartable / fault-tolerant (cluster version). On a cluster, a machine can die mid-job; the job must recover and still finish, not restart from zero.

Explicitly out of scope (name it so you own the scope):

  • Real-time / streaming sort. This is a batch job over a fixed dataset, not a continuously sorted stream. Maintaining a sorted view of live-updating data is a different problem (an index / LSM tree), not a one-shot sort.
  • In-place sort. We are allowed scratch disk space (typically ~2x the input for spill files and output). Sorting 1TB truly in place on disk with no scratch is a needless constraint that no real system accepts.
  • Arbitrary comparators with side effects. The key comparison is a pure, cheap, total-order function on the key bytes. We are not designing for a comparator that does network calls.
  • Sub-second latency. Sorting 1TB is minutes-to-tens-of-minutes work. We optimize throughput (bytes/sec of sequential I/O), not latency of a single record.
  • Compression / encoding format design. We assume a fixed record format. Column encoding and compression matter in practice but are orthogonal to the sort algorithm.

The one decision that drives everything: the data does not fit in memory, so the algorithm must be I/O-bound and sequential, never random-access over the whole dataset. Every hard part below is a consequence of that.

Non-Functional Requirements (NFR)

  • Scale: 1TB input, ~100-byte records => on the order of 10 billion records. Per-machine RAM is 16GB, of which realistically ~10-12GB is usable for sort buffers (the rest is OS, JVM/runtime, I/O buffers). So the usable-RAM-to-data ratio is roughly 1:100 on a single machine.
  • Throughput target: dominated by disk sequential bandwidth. A commodity HDD does ~100-200 MB/s sequential; an SSD does ~500 MB/s to a few GB/s. External sort reads and writes the data a bounded number of times, so wall-clock is roughly (passes * data_size) / disk_bandwidth. Target: single machine in a few hours on HDD or ~1 hour on SSD; a cluster of ~10-20 machines in minutes.
  • Memory ceiling (hard): never allocate a structure proportional to the full dataset in RAM. Every in-memory structure is bounded by a configured buffer size, independent of the 1TB total. This is the invariant that keeps us off the OOM killer.
  • Availability / fault tolerance: single-machine version is a batch job - if the machine dies, restart the job (checkpoint sorted runs so a restart does not redo completed work). Cluster version must tolerate a worker failure by re-running only the failed worker’s tasks, not the whole job.
  • Durability: input is read-only and untouched; output is written durably (fsync / replicated) before the job reports success. Intermediate spill files are scratch and can be lost/recreated on restart.
  • Correctness guarantee: exactly-once semantics on records - count in equals count out, and a final verification pass (or a sampled check) confirms global order.

Back-of-the-Envelope Estimation (BoE)

Real numbers, because they decide the number of passes, the buffer sizes, and the cluster size.

Record and run counts:

Input:            1 TB = 1,000 GB (use 1000 for round numbers)
Record size:      ~100 bytes (10 B key + 90 B value)
Record count:     1 TB / 100 B = ~10,000,000,000 records (10 billion)
Usable RAM/box:   16 GB total, ~10 GB usable for sort buffers
Sorted run size:  ~10 GB (one in-memory sort fills a run)
Number of runs:   1 TB / 10 GB = ~100 sorted runs on a single machine

Single-machine I/O cost (two-phase external sort):

Phase 1 (create runs): read 1 TB, sort in RAM, write 1 TB of runs => 2 TB of I/O
Phase 2 (merge runs):  read 1 TB of runs, merge, write 1 TB output => 2 TB of I/O
Total I/O:             ~4 TB moved (2 read + 2 write) for a 2-pass sort
At 150 MB/s HDD:       4 TB / 150 MB/s = ~7.4 hours (I/O bound)
At 500 MB/s SSD:       4 TB / 500 MB/s = ~2.2 hours
At 2 GB/s NVMe:        4 TB / 2 GB/s  = ~33 minutes
=> The sort is DISK-BANDWIDTH BOUND. Comparisons are free by comparison.

Merge fan-in check (does the merge fit in one pass?):

Runs to merge:       ~100
Merge read buffer:   say 8 MB per run (sequential read efficiency)
RAM for buffers:     100 runs * 8 MB = 800 MB   -> fits easily in 10 GB
=> A single merge pass over 100 runs is feasible. Good - two passes total.
If runs were 100,000 (tiny RAM or huge data), 100k * 8 MB = 800 GB > RAM:
   then merge in multiple rounds (merge groups of ~100, repeat) = more passes.

Distributed sizing (cluster version):

Cluster:             say 20 workers, 16 GB RAM each, local disk each
Data per worker:     1 TB / 20 = 50 GB to own after partitioning
Runs per worker:     50 GB / 10 GB = ~5 sorted runs each -> trivial merge
Network shuffle:     each record moves once to its partition owner => ~1 TB
                     shuffled across the network (minus records already local).
At 10 Gbps/node:     ~1.25 GB/s/node; 1 TB / (20 * 1.25 GB/s) ~ 40 s of pure
                     shuffle if perfectly parallel (real: minutes with overhead).
Wall clock:          minutes, because I/O and network are fully parallelized.

The headline shape: the sort moves ~4x the data volume through disk on one machine (2 read + 2 write), so it is disk-bandwidth bound and the entire game is minimizing passes and keeping every read and write sequential. Distributing it turns the 4TB of serial disk I/O into per-worker chunks that run in parallel, trading disk time for a one-time network shuffle. Number of passes is the metric that matters; buffer sizes and fan-in are chosen to keep it at two.

High-Level Design (HLD)

Start with the single machine, because the distributed design is the same algorithm with a partitioning step bolted on the front. External merge sort has exactly two phases: create sorted runs (the split-sort-spill phase) and merge sorted runs (the k-way merge phase).

                        SINGLE-MACHINE EXTERNAL MERGE SORT

   INPUT 1TB (on disk)                                        OUTPUT 1TB sorted
   ┌───────────────────┐                                      ┌───────────────────┐
   │  unsorted records │                                      │  globally sorted  │
   └─────────┬─────────┘                                      └─────────▲─────────┘
             │ read sequential, 10 GB at a time                         │ write sequential
             ▼                                                          │
   ┌───────────────────────────── PHASE 1: CREATE RUNS ──────────────────────────┐
   │  fill 10 GB buffer -> sort in RAM (quicksort) -> spill sorted run to disk    │
   │  repeat ~100 times                                                           │
   └───────────────────────────────────────┬─────────────────────────────────────┘
                                            │ writes 100 sorted runs to scratch disk
                     ┌──────────┬──────────┬┴─────────┬──────────┐
                     ▼          ▼          ▼          ▼          ▼
                 [run 1]    [run 2]    [run 3]  ...  [run 99]  [run 100]   (each sorted)
                     │          │          │          │          │
                     └──────────┴────┬─────┴──────────┴──────────┘
                                     │ read a small buffer from EACH run
                                     ▼
   ┌───────────────────────────── PHASE 2: K-WAY MERGE ──────────────────────────┐
   │  min-heap over the current head of all 100 runs                             │
   │  pop global-min record -> write to output -> pull next from that run's buffer│
   │  refill a run's buffer from disk when it drains (sequential read)            │
   └──────────────────────────────────────────────────────────────────────────────┘

Single-machine request flow (the batch job, step by step):

  1. Read + fill: read the input sequentially, filling a large in-memory buffer (~10 GB). Stop when the buffer is full (or input ends).
  2. Sort in RAM: sort the buffer with a fast in-memory sort (quicksort / introsort / radix sort on the fixed-width key). This is O(n log n) comparisons but pure CPU/RAM, cheap relative to disk.
  3. Spill: write the sorted buffer to a new run file on scratch disk, sequentially. Clear the buffer.
  4. Repeat 1-3 until the whole input is consumed. Now scratch disk holds ~100 sorted run files, each internally sorted, collectively covering all records.
  5. Merge: open all 100 runs, read a small buffer from each, and run a k-way merge using a min-heap keyed on the head record of each run. Repeatedly pop the global minimum, append it to the output buffer, and advance that run. When a run’s read buffer empties, refill it sequentially from disk. When the output buffer fills, flush it sequentially to the output file.
  6. Done: every record has flowed input -> run -> merged output exactly once. The output is globally sorted. Verify count and (optionally) order.

Now the distributed version. The key idea that makes a cluster produce globally sorted output (not just per-machine sorted): range partition first. Decide value ranges up front so that worker 0 owns [-inf, a), worker 1 owns [a, b), and so on, with ranges chosen so each worker gets ~equal data. Route every record to its range owner, then each worker external-sorts only its own range locally. Concatenating workers’ outputs in range order gives a globally sorted file, with no final cross-machine merge needed.

                         DISTRIBUTED (TeraSort-STYLE) SORT

   INPUT 1TB (sharded across storage)
   ┌────────┬────────┬────────┬────────┐
   │ block0 │ block1 │  ...   │ blockN │   (HDFS / S3 / distributed FS)
   └───┬────┴───┬────┴───┬────┴───┬────┘
       │        │        │        │
   ┌───▼────────▼────────▼────────▼───── STAGE 0: SAMPLE + RANGE BOUNDARIES ────┐
   │  coordinator samples ~keys, picks R-1 split points so each range ~= 1/R     │
   │  of the data  =>  range map:  r0=[-inf,a)  r1=[a,b) ... r(R-1)=[z,+inf)      │
   └───────────────────────────────┬────────────────────────────────────────────┘
                                    │ broadcast range map to all mappers
       ┌────────────┬──────────────┼──────────────┬────────────┐
       ▼            ▼              ▼              ▼            ▼
   ┌───────┐    ┌───────┐      ┌───────┐      ┌───────┐    ┌───────┐
   │MAPPER │    │MAPPER │      │MAPPER │ ...  │MAPPER │    │MAPPER │   STAGE 1: MAP
   │read + │    │read + │      │read + │      │read + │    │read + │   partition each
   │route  │    │route  │      │route  │      │route  │    │route  │   record by range
   └───┬───┘    └───┬───┘      └───┬───┘      └───┬───┘    └───┬───┘
       └── shuffle: every record sent to the reducer that owns its range ──┐
                                    │ (network, ~1 TB total)                │
       ┌────────────┬──────────────┼──────────────┬────────────┐           │
       ▼            ▼              ▼              ▼            ▼            │
   ┌───────┐    ┌───────┐      ┌───────┐      ┌───────┐    ┌───────┐        │
   │REDUCER│    │REDUCER│      │REDUCER│ ...  │REDUCER│    │REDUCER│  STAGE 2: REDUCE
   │r0     │    │r1     │      │r2     │      │r(R-2) │    │r(R-1) │  external-sort
   │local  │    │local  │      │local  │      │local  │    │local  │  own range only
   │extsort│    │extsort│      │extsort│      │extsort│    │extsort│
   └───┬───┘    └───┬───┘      └───┬───┘      └───┬───┘    └───┬───┘
       ▼            ▼              ▼              ▼            ▼
   [part-00000] [part-00001]   [part-00002]  [part-000..] [part-R-1]  STAGE 3: OUTPUT
        \___________\____________ concatenate in range order = GLOBAL SORT ___/

Distributed flow, step by step:

  1. Sample (stage 0): the coordinator reads a small random sample of keys (a few hundred thousand keys is plenty for 10B records) and picks R-1 split points so each of the R ranges holds ~1/R of the data. This range map is the crux - get it wrong and one reducer gets swamped (deep dive 2).
  2. Map (stage 1): each mapper reads its input blocks sequentially, and for each record computes which range it belongs to (binary search over the R-1 split points) and buffers it for the reducer that owns that range.
  3. Shuffle: mappers send each record to its owning reducer over the network. This is the ~1TB data movement. Records for the same reducer are batched into large network buffers for throughput.
  4. Reduce (stage 2): each reducer receives all records in its range - roughly 50GB on a 20-node cluster - which does not fit in its 16GB RAM, so the reducer runs the single-machine external merge sort above on just its range.
  5. Output (stage 3): each reducer writes one sorted output file (part-00000, part-00001, …). Because reducer i owns a strictly-lower range than reducer i+1, concatenating the parts in order yields a globally sorted 1TB file. No final merge across machines.

The structural insight: external merge sort turns an impossible random-access problem into two sequential streaming passes over disk, and distributing it means doing the ordering decision once, up front, via range partitioning - so each machine sorts an independent slice and the slices already line up globally. The single-machine external sort is the reducer; the distributed sort is external sort with a sampling-driven partition on the front.

Component Deep Dive

Four hard parts, naive-first then evolved: (1) sorting data bigger than RAM at all; (2) picking range boundaries so no reducer is overloaded (data skew); (3) making the k-way merge fast enough to be disk-bound not CPU-bound; (4) fault tolerance so a dead worker does not restart the whole job.

1. Sorting more data than RAM: load-and-sort -> spill sorted runs, then merge

The core problem is that the dataset is 64x bigger than memory. Every in-memory sort assumes it can index any element at any time.

Naive approach: read it all into an array and call sort()

Load the file into a giant array (or memory-map it and sort in place) and hand it to the standard library sort.

data = read_entire_file("input_1tb")   # allocate 1 TB in RAM
sort(data)                             # quicksort over the whole array
write(data, "output")

Where it breaks:

  • It cannot allocate 1TB of RAM on a 16GB machine. The allocation fails, or on a machine with swap the OS starts paging and the “sort” becomes billions of random 4KB page faults against disk. Quicksort’s access pattern is effectively random over the array; backed by disk swap, every comparison can trigger a seek. On an HDD at ~10ms/seek and 10 billion records doing O(log n) ~= 33 accesses each, you are looking at geological time. It does not just get slow, it never finishes.
  • Memory-mapping does not save it. mmap lets you address 1TB, but quicksort’s random partitioning still touches pages all over the file, so the page cache thrashes exactly the same way. Addressability is not the problem; the random access pattern is.

Any algorithm that random-accesses the full dataset is dead the instant the dataset exceeds RAM. The fix is to never random-access more than a bufferful at once.

Evolved approach: two-phase external merge sort (sorted runs + k-way merge)

Bound every in-memory structure by a fixed buffer size, independent of total data. Phase 1 reads the input in RAM-sized chunks, sorts each chunk in memory (random access confined to a 10GB buffer, which is fine), and spills it as a sorted run. Phase 2 merges the runs with a streaming k-way merge that only holds one small buffer per run.

# PHASE 1: create sorted runs
run_id = 0
while not eof(input):
    buf = read_chunk(input, size = 10 GB)     # sequential read
    in_memory_sort(buf)                        # quicksort/radix - RAM-only, fast
    write_run(f"run_{run_id}", buf)            # sequential write, run is sorted
    run_id += 1
# now ~100 sorted runs on scratch disk

# PHASE 2: k-way merge with a min-heap
readers = [buffered_reader(f"run_{i}") for i in range(run_id)]   # small buf each
heap = min_heap()
for i, r in enumerate(readers):
    heap.push( (r.peek_key(), i) )             # seed with each run's head
out = buffered_writer("output")
while heap not empty:
    key, i = heap.pop_min()                    # global minimum among run heads
    out.write( readers[i].next() )             # emit it (sequential write)
    if readers[i].has_more():
        heap.push( (readers[i].peek_key(), i) )# advance that run
out.flush(); fsync()

Why this is right:

  • Memory is bounded by buffer size, not data size. Phase 1 holds one 10GB buffer; phase 2 holds ~100 small read buffers plus one output buffer - a few hundred MB total. Neither scales with the 1TB total, so it fits in 16GB with room to spare. This is the invariant that makes the impossible possible.
  • Every disk access is sequential. Phase 1 reads the input straight through and writes each run straight through. Phase 2 reads each run sequentially (the small buffer amortizes seeks) and writes the output sequentially. There is no random disk access anywhere, so the job runs at disk bandwidth, not disk seek rate - the difference between hours and never.
  • Two passes is provably enough here. With ~100 runs and a merge fan-in of 100, one merge pass suffices, so total I/O is 2 reads + 2 writes = ~4TB. The number of passes is 1 + ceil(log_fanin(num_runs)); keeping fan-in >= num_runs keeps it at two. If runs vastly outnumbered the fan-in the merge would take multiple rounds (merge groups, then merge the merges), adding a pass each round - the cost you minimize.
  • The in-memory sort choice is a free lever. Since keys are fixed-width (10 bytes), phase 1 can use radix sort (O(n) in the key width) instead of comparison sort, cutting CPU further - though CPU is already not the bottleneck. Replacement selection can also produce runs ~2x RAM size on average (a heap that keeps emitting the smallest key >= last-emitted), halving the run count and sometimes saving a pass.

External merge sort is the whole answer to “bigger than RAM”: confine random access to a bufferful, stream everything else sequentially, and pay for it in a bounded number of passes over disk.

2. Range boundaries and data skew: fixed ranges -> sample-driven balanced ranges

In the distributed version, global order comes from range partitioning: reducer i owns a value range, and the parts concatenate in order. The hard part is choosing the range boundaries so every reducer gets roughly equal work.

Naive approach: split the key space into equal-width ranges

If keys are 10-byte values, split the value space uniformly: reducer 0 gets keys 0x00... to 0x19..., reducer 1 gets 0x1A... to 0x33..., and so on - equal-width slices of the key domain.

range_width = KEY_SPACE / R
boundary[i] = i * range_width          # equal slices of the key domain
route(record) = record.key / range_width

Where it breaks:

  • Real data is not uniformly distributed, so ranges are wildly unbalanced. If keys are usernames, most start with common letters and almost none start with z or a digit; if keys are timestamps clustered in business hours, whole ranges are near-empty while others are packed. One reducer gets 300GB while another gets 5GB. The 300GB reducer becomes the straggler that the entire job waits on - a 20-node cluster runs at the speed of its slowest node, so skew destroys the parallelism you paid for.
  • Worst case, one reducer gets everything. If all keys share a prefix (e.g. all IDs start with the same epoch prefix), equal-width partitioning routes ~all data to one reducer, which then has to external-sort the entire 1TB alone - you have distributed nothing.

Equal-width ranges assume a uniform key distribution that real data never has. You must partition by the data’s actual distribution, not the key domain’s geometry.

Evolved approach: sample the keys, pick boundaries at the sampled quantiles

Before mapping, read a random sample of keys, sort the sample, and pick the R-1 split points at the sample’s quantiles - the split that puts 1/R of the sample in each range puts ~1/R of the actual data in each range, whatever the distribution. This is exactly what Hadoop’s TeraSort InputSampler does.

# STAGE 0: sample-based range map
sample = []
for block in random_subset(input_blocks):
    sample += read_some_keys(block, k per block)      # ~100k-1M keys total
sort(sample)
R = num_reducers
boundary = [ sample[ (i * len(sample)) / R ] for i in 1..R-1 ]   # quantile cuts
# boundary now has R-1 split points; range i = [boundary[i-1], boundary[i])

# routing at map time (binary search over R-1 boundaries)
def route(key):
    return binary_search_range(boundary, key)         # O(log R), returns reducer id

Why this is right:

  • Balanced by construction, for any distribution. Quantiles of a representative sample track quantiles of the population, so each reducer gets ~1/R of the data regardless of how clumped the keys are. Skewed usernames, clustered timestamps, prefixed IDs - all balance out because we cut where the data actually is, not where the domain is. No straggler, so the cluster runs at full parallel speed.
  • A small sample is statistically sufficient. Sorting 10 billion records but sampling only ~100k-1M keys is enough to place R-1 boundaries accurately (sampling error on a quantile shrinks as 1/sqrt(sample_size)). The sample sort is trivial (fits in RAM on the coordinator), so stage 0 is cheap - seconds - and it saves the whole job from skew.
  • Routing stays O(log R) per record. With the sorted boundary array, each mapper binary-searches the R-1 split points to find a record’s reducer - a handful of comparisons per record, negligible next to I/O. The range map is small (R-1 keys) and broadcast to every mapper.
  • Handles residual hot keys explicitly. If a single key value is so frequent it alone exceeds one reducer’s capacity (a true heavy hitter, e.g. a null or default key that is 20% of the data), sampling reveals it and you special-case it: give that one key its own reducer, or split its records across several reducers with a secondary sort (they are all equal on the key anyway, so any split preserves order). Naming this is the difference between a design that handles real data and one that assumes clean data.

Sample-driven quantile boundaries are what make the distributed sort actually distribute: they turn an adversarial, skewed key distribution into R evenly-loaded reducers, which is the only way the cluster finishes in minutes instead of waiting on one overloaded node.

3. The k-way merge: rescan all runs -> min-heap (loser tree) with big sequential buffers

Phase 2 merges ~100 sorted runs into one stream. It reads and writes the entire dataset, so it must run at disk bandwidth, and it must pick the global minimum across 100 runs efficiently.

Naive approach: linear scan of all run heads to find the minimum

To emit the next record, look at the current head of all 100 runs and pick the smallest by scanning them all.

while any run has records:
    min_i = argmin over all 100 runs of run[i].head_key    # O(k) scan per record
    emit(run[min_i].next())

Where it breaks:

  • O(k) per record makes the merge CPU-bound at high k. Scanning 100 run heads for each of 10 billion records is 10^12 comparisons. At a few ns each that is thousands of seconds of pure comparison - and with a larger fan-in (merging thousands of runs in a multi-round merge) it gets quadratically worse. The merge stops being disk-bound and becomes comparison-bound, wasting disk bandwidth that sits idle.
  • Tiny per-run reads cause seek thrash. If each run is read one record at a time straight off disk, the 100 runs interleave into a random seek pattern across the disk - 100 files being read in a rotating fashion is effectively random I/O on an HDD, collapsing throughput to seek rate.

A linear scan wastes CPU, and unbuffered per-record reads waste the disk. Both must be fixed.

Evolved approach: min-heap (or loser tree) over run heads, with large per-run read buffers

Keep a min-heap of size k keyed on each run’s current head. Popping the minimum and pushing the run’s next record is O(log k) instead of O(k). And give each run a large sequential read buffer (e.g. 8-16MB) so disk reads are big sequential gulps, not per-record seeks. Symmetrically, buffer the output and flush in large sequential writes.

heap = min_heap()                     # size k, keyed on head record's key
buffers = [ prefetch(run_i, 8 MB) for i in 1..k ]   # big sequential reads
for i in 1..k: heap.push((buffers[i].head_key(), i))
out = write_buffer(16 MB)

while heap not empty:
    key, i = heap.pop()               # O(log k) to get global min
    out.append(buffers[i].pop())      # emit record; flush out when full (sequential)
    if buffers[i].empty():
        buffers[i].refill(8 MB from run_i)   # one big sequential read, amortized
    if buffers[i].has_data():
        heap.push((buffers[i].head_key(), i))
out.flush(); fsync()

Why this is right:

  • O(log k) selection keeps the merge disk-bound. A heap picks the minimum in log(100) ~= 7 comparisons per record instead of 100, cutting merge CPU by ~14x and ensuring the CPU can always keep the disk fed. A loser tree (a tournament tree, the classic external-sort structure) does even better - one comparison per level, ~7 per record, with better cache behavior and fewer swaps than a binary heap. Either way, selection is no longer the bottleneck.
  • Large per-run buffers convert random I/O into sequential I/O. Reading 8-16MB from a run at a time means each run is read as a sequence of large sequential chunks; the disk head does ~one seek per 8MB instead of one seek per 100 bytes. This is the single most important merge tuning knob - it is why the buffer size in the BoE mattered. Total buffer RAM = k * buffer_size (100 * 8MB = 800MB), comfortably within budget.
  • Fan-in is bounded by RAM, and that bounds passes. The max fan-in is usable_RAM / per_run_buffer. With 10GB and 8MB buffers that is ~1250-way merge - far more than our 100 runs, so one pass. Only if runs exceeded the fan-in would you do a multi-round merge (merge into fewer, bigger runs, repeat), and each round is another full read+write of the data - so you size buffers to keep fan-in high and passes low.
  • Double-buffering hides I/O latency. While the merge consumes one buffer per run, a background thread prefetches the next chunk into a second buffer, so the CPU never stalls waiting on a disk read. This overlaps compute and I/O and pushes the merge to true disk-bandwidth throughput.

A heap/loser-tree merge with big sequential buffers is what makes phase 2 run at the disk’s bandwidth limit: cheap minimum-selection so the CPU keeps up, and large sequential reads/writes so the disk is never seeking.

4. Fault tolerance: restart from zero -> checkpoint runs and re-run only failed tasks

A 1TB sort on a cluster runs for minutes to tens of minutes across many machines; over that window a machine failing is normal, not exceptional. The job must survive it.

Naive approach: hold everything in memory and pipe map straight to reduce; restart on any failure

Run all mappers and reducers as one pipeline, streaming records mapper->reducer in memory, and if anything dies, start the whole job over.

pipeline: mappers stream records -> reducers sort in memory -> write output
on any worker crash: kill job, restart from the original 1 TB input

Where it breaks:

  • Restarting from zero wastes all completed work. If a reducer dies 90% through a 20-minute job, you have re-read and re-shuffled the entire 1TB and thrown away 18 minutes of every other machine’s work. With commodity hardware failing regularly, a big job might never complete because it keeps restarting before it finishes - the failure probability over the full run approaches 1 as the cluster grows.
  • In-memory reducers cannot recover their state. A reducer accumulating its 50GB range in RAM (impossible anyway - it exceeds 16GB) has no durable checkpoint; its partial sorted state vanishes on crash and must be rebuilt from scratch by re-shuffling.

All-or-nothing recovery does not survive commodity hardware at scale. Recovery must be per-task, from durable intermediate state.

Evolved approach: materialize map output to disk, re-run only failed tasks, speculative execution for stragglers

Make the boundary between map and reduce durable: each mapper writes its partitioned output to local disk (grouped by reducer), and only after a mapper’s output is committed is it “done.” Reducers pull committed map outputs. If a worker dies, the scheduler re-runs just that worker’s tasks on another machine, reading the durable inputs those tasks need - the rest of the job is untouched. This is the MapReduce fault model.

# durable map output (the recovery boundary)
mapper m:
    for record in input_split_m:
        r = route(record.key)                 # which reducer/range
        spill[r].append(record)               # buffer per reducer
    for r in 1..R:
        write_local(f"map_{m}_part_{r}", sort(spill[r]))   # committed to disk
    mark_task_done(m)                          # scheduler records completion

# reduce pulls committed map outputs, external-sorts, writes final part
reducer r:
    inputs = [ fetch(f"map_{m}_part_{r}") for m in all_completed_mappers ]
    external_merge_sort(inputs) -> write(f"part_{r:05d}")   # durable output
    mark_task_done(r)

# scheduler recovery
on worker_death(w):
    for task in tasks_on(w):
        if task.output_committed: skip         # its result survives on disk
        else: reschedule(task) on a live worker # re-run ONLY that task
# straggler mitigation
if a task runs > 1.5x median: launch a speculative duplicate on another node;
    take whichever finishes first (idempotent - task output is deterministic).

Why this is right:

  • Recovery is per-task, not per-job. Because map output is durable, a dead reducer’s replacement re-reads the same committed map partitions and redoes only that one range; a dead mapper’s split is re-read and re-partitioned alone. The other 19 machines keep their progress. Job completion time degrades gracefully with failures instead of collapsing to “never.”
  • The map/reduce disk boundary is the checkpoint. Materializing shuffle output to disk costs one extra write of the data, but it is what makes the job restartable - a deliberate durability-for-recovery trade. (Systems like Spark can keep this in memory for speed and recompute lost partitions from lineage instead; the trade is memory/recompute vs disk checkpoint. For a pure batch sort on commodity hardware, disk materialization is the safe default.)
  • Task output is deterministic and idempotent, so re-running is safe. Sorting is a pure function of its input, so a re-run (or a speculative duplicate) produces byte-identical output. That is what lets the scheduler freely re-execute tasks and dedup by “first to commit wins” without any risk of corrupting the result - exactly-once output falls out of deterministic tasks plus commit-on-completion.
  • Speculative execution kills stragglers. A single slow disk or a partially-failing node can make one task lag and stall the whole job. Launching a backup copy of any task running far past the median, and taking the first to finish, caps the job on the fast majority instead of the slow tail - the same defense as the sampling that prevents data-skew stragglers, applied to hardware skew.

Durable map output plus per-task re-execution plus speculative backups is what makes a multi-machine sort actually finish on unreliable commodity hardware: failures cost one task’s work, not the whole job, and deterministic tasks make re-running always safe.

API Design & Data Schema

This is a batch job, so the “API” is a job-submission control plane plus the on-disk formats, not a request/response service.

Job control API

POST /sortjobs                      # submit a sort job
  { "input": "s3://bucket/data/*",  "output": "s3://bucket/sorted/",
    "key": { "offset": 0, "length": 10, "type": "bytes", "order": "asc" },
    "record_bytes": 100, "reducers": 20, "sample_size": 1000000,
    "stable": false }
  -> 202 { "job_id": "sort-7f3a", "state": "SAMPLING" }

GET  /sortjobs/{job_id}             # poll status
  -> { "job_id": "sort-7f3a", "state": "REDUCING",
       "map_done": 512, "map_total": 512, "reduce_done": 14, "reduce_total": 20,
       "bytes_shuffled": 1_010_000_000_000, "stragglers": 1 }

GET  /sortjobs/{job_id}/verify      # post-run correctness check
  -> { "input_records": 10_000_000_000, "output_records": 10_000_000_000,
       "order_ok": true, "boundary_ok": true }     # counts match, order verified

DELETE /sortjobs/{job_id}           # cancel; cleans up scratch runs

On-disk formats

Input / output record (fixed-width, the TeraSort shape):

record: [ key: 10 bytes ][ value: 90 bytes ]      # 100 bytes, no delimiters needed
   fixed width => record N starts at offset N*100; no parsing, pure byte math.
   output: same records, globally ordered by key ascending.

Sorted run file (phase-1 spill, single-machine or reducer-local):

run_{id}: concatenation of sorted fixed-width records, header:
   { magic, record_bytes, record_count, min_key, max_key }   # min/max aid merge
   body: record_count records, sorted asc, written sequentially.

Map output partition (the durable shuffle boundary):

map_{m}_part_{r}: records from mapper m destined for reducer r,
   pre-sorted within the partition (so the reducer merges pre-sorted inputs).
   { header: mapper_id, reducer_id, record_count } + sorted records

Range map (stage-0 output, broadcast to all mappers):

{ reducer_count: R,
  boundaries: [ key_1, key_2, ..., key_{R-1} ],   # R-1 split keys, ascending
  hot_keys: [ { key, dedicated_reducers:[..] } ] } # explicit heavy-hitter handling

Job metadata (coordinator, strongly consistent - small):

job: { job_id, input, output, state, range_map_ref,
       map_tasks: [ {id, worker, state, output_ref} ],
       reduce_tasks: [ {id, range, worker, state, output_ref} ] }

SQL vs NoSQL vs neither. The dataset itself lives in flat files on a distributed filesystem or object store (HDFS / S3), not a database - a database would add per-row overhead, indexing, and a query layer we do not want for a one-shot bulk sort of fixed-width records; raw sequential file I/O is strictly faster and simpler for this workload. The only thing that needs a real store is the small, strongly-consistent job/coordinator metadata (task states, range map, worker liveness), which fits a Raft-backed metadata store or a coordinator’s replicated log. So: files for the bulk data, a small consistent store for control state. No general-purpose database on the hot path.

Bottlenecks & Scaling

Where it breaks first, in order, and the fix for each.

1. Single-machine disk bandwidth is the hard floor. On one box the job moves ~4TB through one disk; at 150 MB/s that is ~7 hours and no amount of CPU helps. Fix: stripe across multiple disks (RAID-0 / JBOD with parallel spill) so phase-1 runs and phase-2 buffers hit several spindles at once, multiplying effective bandwidth; or distribute across machines so each moves only 1/N of the data on its own disk in parallel.

2. Too many runs forces extra merge passes. If RAM is small or data is huge, the run count exceeds the merge fan-in, so the merge needs multiple rounds - each round is another full read+write (a whole extra pass over 1TB). Fix: maximize run size (use all usable RAM per run; replacement selection for ~2x runs) to cut run count, and maximize fan-in with a bounded per-run buffer so one merge pass covers all runs. Passes are the cost function; keep it at two.

3. Data skew creates a straggler reducer. One range gets far more than 1/R of the data and the whole cluster waits on it. Fix: sample-driven quantile boundaries (deep dive 2) so ranges are balanced for the actual distribution; dedicated or split reducers for heavy-hitter keys; monitor per-reducer input bytes and rebalance.

4. Shuffle network is the cluster bottleneck. Moving ~1TB across the network in stage 1 can dominate wall-clock if the network is thin or the pattern is all-to-all bursty. Fix: large batched shuffle writes (not per-record), combine/pre-sort map output per partition so reducers merge pre-sorted streams, rack-aware placement to keep traffic local where possible, and overlap shuffle with map (start sending completed partitions before a mapper finishes its whole split).

5. Stragglers from slow hardware. A single slow disk or degraded node makes one task lag and stalls the job. Fix: speculative execution (deep dive 4) - duplicate a task running past 1.5x the median and take the first to finish; deterministic task output makes this safe.

6. Small I/O / random seeks anywhere. Per-record reads on the merge, or tiny shuffle messages, collapse throughput to seek rate. Fix: large sequential buffers everywhere - 8-16MB per run on merge, MB-scale shuffle batches, big output writes. Sequential I/O is the entire performance model; any random access is a bug to hunt down.

7. Coordinator as a single point of failure. The scheduler holding task state, the range map, and worker liveness must not be one fragile node. Fix: run coordinator state on a Raft quorum (3/5 nodes); keep its state tiny (task metadata only, never records) so it stays fast and highly available; workers heartbeat and the coordinator reassigns tasks of dead workers.

8. Output durability and verification. A “finished” job that silently dropped or duplicated records is worse than a failed one. Fix: fsync/replicate output before reporting success; run a verification pass - record count in == count out, and a sampled or full scan asserting each part is sorted and each part’s max key < the next part’s min key (global order across boundaries). The boundary_ok check is what proves the concatenation is truly globally sorted.

Shard key, stated plainly: the distributed sort shards by key range, with boundaries chosen from a random sample so each range holds ~1/R of the data. The range is simultaneously the unit of parallelism (one reducer per range), the unit that produces global order (ranges concatenate in order with no cross-machine merge), and the thing skew attacks (a bad boundary overloads one reducer) - so sampling the key distribution to place balanced boundaries is the single decision that makes the whole distributed sort both correct and fast.

Wrap-Up

The trade-offs that define this design:

  • Sequential I/O over random access, always. The dataset is 64x RAM, so any algorithm that random-accesses the whole thing is dead. External merge sort confines random access to a RAM-sized buffer and streams everything else sequentially, trading a bounded number of full passes over disk (two, here) for the ability to sort data far larger than memory. Passes over disk is the cost function you minimize.
  • Range partition up front over merge at the end. The distributed version decides ordering once, via sampled range boundaries, so each machine sorts an independent slice whose outputs already concatenate into global order - no expensive final cross-machine merge. The price is a one-time ~1TB network shuffle and total dependence on a good, sample-driven partition to avoid skew.
  • Balanced by sampling, not by assumption. Equal-width ranges assume a uniform key distribution real data never has; quantile boundaries from a cheap sample balance any distribution, which is the difference between a 20x speedup and one overloaded straggler doing all the work.
  • Durable intermediate state over restart-from-zero. Materializing map output to disk costs one extra write but makes recovery per-task; combined with deterministic, idempotent tasks and speculative execution, the job finishes on commodity hardware that fails mid-run rather than restarting forever.
  • Files and a tiny consistent control store over a database. Bulk data is raw sequential files on a distributed FS/object store (no per-row overhead); only the small task/range metadata needs a strongly-consistent store. The right tool for a one-shot bulk sort is not a database.

One-line summary: sort 1TB on 16GB machines with external merge sort - split the data into RAM-sized sorted runs spilled to disk, then k-way merge them with a heap and big sequential buffers in two disk passes - and distribute it by sampling the keys to pick balanced range boundaries so each worker external-sorts one range and the ranges concatenate into a globally sorted file, with durable map output and speculative execution keeping the job alive on failing commodity hardware.