Sort 1 TB on 1,000 machines with 1.5 GB RAM each

01The design, in the six-step framework

1. Requirements

Functional.

• Input: 1 TB of fixed-size records (say 10 billion records of 100 bytes, each with a 10-byte key) spread over a distributed file system.
• Output: the same records in global key order, written as a set of files such that file i holds keys smaller than every key in file i+1.
• Resources: 1,000 machines with 1.5 GB RAM each, local disks, a 1 Gbit/s network link each, and a shared file system.
• Out of scope: a general query engine; this is one job, run well.

Non-functional.

• Finish in minutes, not hours; a target of under 10 minutes wall clock.
• Correct despite machine failures during the run; a slow machine must not make the job take twice as long.
• Deterministic output: the same input produces byte-identical output files.

2. Core entities

• Input split: file, byte range, assigned mapper, state.
• Key sample: a key drawn from a split; the sample set determines partition boundaries.
• Partition: index i, key range [lo_i, hi_i), assigned reducer.
• Spill file: mapper, partition, sorted run on local disk, size.
• Task attempt: task id, attempt number, machine, state, progress; the unit of retry.

3. API

This is a batch job, so the "API" is the job specification and the coordinator's internal RPCs.

• submit(input_path, output_path, num_partitions=1000) → job id; status(job_id) → phase and per-task progress.
• Coordinator → worker: sample(split) → keys; map(split, boundaries) → spill locations; reduce(partition, spill locations) → output file.
• Worker → coordinator: heartbeat(task, progress) every few seconds; the coordinator uses it for failure detection and straggler detection.
• Workers serve spill files to each other over HTTP ranges: GET /spill/{task}/{partition}.

4. High-level design

The job, phase by phase:

1. Sample. Each machine reads a few thousand random keys from its split and sends them to the coordinator, which sorts the 10 million samples and picks 999 boundary keys at equal quantiles, then broadcasts them.
2. Map. Each machine streams its 1 GB, assigns each record to a partition by binary search on the boundaries, buffers per partition, sorts each buffer in memory when it fills, and spills sorted runs to local disk; at the end it merges its runs into one sorted spill per partition.
3. Shuffle. Reducer i fetches partition i's spill from all 1,000 mappers over the network; every machine is simultaneously a server and a client.
4. Reduce. Reducer i performs a 1,000-way merge of sorted streams (a heap of 1,000 heads) and writes output file i directly to the shared file system.
5. Output file i contains partition i in order; because boundaries are global, the concatenation of files 0..999 is the fully sorted data.

5. Deep dives

Why sample first, and what goes wrong without it?

If partitions were chosen by hashing or by fixed key ranges, a skewed key distribution (lots of keys starting with "a") would send 10 GB to one reducer, which has 1.5 GB of RAM and would run for ten times longer than the rest. Sampling estimates the distribution, so every reducer receives about 1 GB. With 10,000 samples per machine the quantile error is small enough that the largest partition is within a few percent of the mean. The job cannot start mapping until the boundaries exist, so the sampling pass is a short extra read; it is worth it every time. This is range partitioning with data-driven boundaries, the same idea a database uses to split a sorted table.

How does a machine sort 1 GB in 1.5 GB of RAM?

It does not sort it all at once. It fills a buffer of about 500 MB, sorts it (quicksort or radix on the 10-byte key), writes the sorted run to disk, and repeats; then it does a k-way merge of the runs, which needs only a block of each run in memory. The map side spills per partition, so a run is really 1,000 small sorted segments with an index; the merge on the reduce side reads 1,000 remote segments and needs 1,000 × 64 KB = 64 MB of read buffers, well within budget. The whole design is external merge sort with the merge spread across machines; recognising that in one sentence is what the interviewer is listening for.

What about the slow machine and the dead machine?

A dead mapper is detected by missed heartbeats and its split is rerun elsewhere; because map output is on local disk, a dead mapper after finishing still loses its spills, so its split is rerun too, which is why map tasks are small and idempotent. A dead reducer is rerun by refetching from the mappers, which keep spills until the job commits. The slow machine is the subtler case: one machine at half speed makes the shuffle and the merge wait for it. The coordinator watches progress rates and, when a task is well behind the median with most tasks done, launches a speculative duplicate on an idle machine; whichever finishes first wins and the other is killed. Output is written to a temporary name and renamed on success, so two attempts never produce a half-written file. These are the same three answers as in the original MapReduce paper, and naming them as standard batch-processing mechanics is fine.

Why might the shuffle be slower than the arithmetic says?

Incast: 1,000 mappers sending to one reducer at once overflow the switch buffer at that reducer's port, packets drop, TCP backs off, and the link runs far below its rate. Mitigations are to fetch from a bounded number of sources at a time (say 20), to randomise fetch order so the whole fleet does not hit the same mapper, and to start fetching as soon as some mappers finish instead of waiting for all. If the cluster's bisection bandwidth is lower than the sum of the links, the shuffle time is bandwidth divided by the bisection, not by the per-link rate; ask about the topology before promising a number.

6. What a strong answer sounds like

• Say "external merge sort, distributed" in the first minute, then do the per-machine arithmetic for memory, disk and network.
• Explain sampling for balanced range partitions and what skew does without it.
• Name the three failure handlers: rerun small idempotent tasks, speculative execution for stragglers, and rename-on-commit for output.
• Mention incast and bisection bandwidth as the reasons the shuffle estimate is a lower bound.
• Expected follow-ups: what if the keys are not uniform in size; what if the input is 100 TB (more waves of tasks, and the merge fan-in gets large); how to verify the output (per-file key ranges and a checksum of counts).
• Common mistake: streaming everything through one coordinator, or assuming 1 GB fits in 1.5 GB with room to sort.

02Where this question was reported

• interviewkickstart.com · Practice question
• interviewkickstart.com · Listed verbatim.
• Exponent / Aced · L4 candidate hit distributed merge sort inside a coding round.