Distributed Sort

TL;DR

Use external merge sort as the primitive and distribute it by sampled key ranges. A coordinator samples the input, chooses approximately even partition boundaries, and broadcasts them to map workers. Map workers read sequential input splits, sort memory-sized chunks into durable sorted runs, and route records to range owners. Reduce workers merge the runs for each range and publish atomically. Durable task state, shared run storage, heartbeats, retries, and task-scoped output paths limit failures to individual tasks.

Scope and assumptions

• This is a bounded batch sort on distributed storage such as HDFS or object storage. Streaming, sub-second latency, secondary sort keys, and custom comparators are below the line.
• The baseline is illustrative: 10 TB in under 2 hours on 100 nodes, with 128 GB RAM and 10 Gbps per node. A 128 GB logical sort chunk is shown for simple arithmetic; production workers must reserve memory for the runtime, buffers, and merge state.
• Partition boundaries are range-based, not hash-based, so concatenating output partitions in boundary order yields a global sort. Sampling error and duplicate/hot keys need explicit handling.
• Intermediate runs are durable before a task is marked complete. HDFS can use atomic rename; object storage needs conditional publication or an equivalent immutable-manifest protocol because object keys do not generally support native rename.
• Throughput, sample rate, buffer sizes, fan-in, failure rates, and node counts are workload and hardware assumptions. Benchmark them rather than treating the figures below as universal guarantees.

The calculations use approximate decimal units unless stated otherwise. The goal is to expose the dominant terms—sequential I/O, shuffle bandwidth, skew, and merge passes.

What is a distributed sorting system?

A distributed sorting system takes a dataset far too large for any single machine and produces a globally sorted output across a cluster of workers. The engineering challenge is that RAM is orders of magnitude smaller than the input: you must spill sorted chunks to disk repeatedly, then merge them back, while minimizing the number of expensive disk passes.

This question forces reasoning across the full hardware stack, from RAM to SSD to network bandwidth. It also tests how a MapReduce or TeraSort-style pipeline turns an apparently serial problem into a parallel one.

Functional Requirements

Core Requirements

1. Sort a dataset too large to fit in the RAM of any single machine.
2. Output a globally sorted sequence, split into N partition files on distributed storage.
3. Accept a configurable sort key (field name or byte offset).
4. Tolerate any single worker failure without restarting the full sort.

Below the Line (out of scope)

• Streaming sort of infinite data (use a Top-K heap service instead).
• Real-time sort with latency under a second.
• Secondary sort keys and custom comparators.

The scope is bounded datasets on distributed storage (HDFS, S3). Streaming and real-time variants branch off at the very first design decision and become fundamentally different systems. Custom comparators are a useful extension but do not change the core architecture, so we defer them.

Call out the scope boundary explicitly. Streaming and batch sorting have different state, latency, and backpressure requirements; combining them without a clear contract obscures both designs.

Hardest part: Partitioning the keyspace evenly across workers. If your 100-worker job sends 40% of records to one worker because your partition boundaries are poorly chosen, that worker determines the job's end time regardless of how fast everything else runs. Partition-boundary selection from a representative sample is what separates a working sort from a fast sort.

Non-Functional Requirements

Performance

• Sort 10TB in under 2 hours on a 100-node cluster with 128GB RAM and 10Gbps per node.
• Each worker must sustain near-peak sequential I/O throughput (target: above 80% of hardware max).
• Minimize disk passes: with an illustrative 100GB per node and 1GB/s sequential I/O, an additional full read-plus-write pass adds roughly 3.3 minutes to each node's work (about 1.7 minutes for a read alone). Extra passes compound across the merge tree.

Reliability

• Tolerate any single worker failure without restarting the full job; re-run only the failed tasks.
• Coordinator failure must be recoverable from checkpointed state within 5 minutes.

Scalability

• Scale linearly from 10TB to 1PB by adding commodity nodes to the cluster.
• Partition count is configurable; default is one output partition per reduce worker.

Consistency

• Output partitions are globally sorted with no overlapping key ranges and no gaps.
• Partition boundaries are stored alongside the output so consumers can seek without scanning all partitions.

Read/write ratio that shapes this design: External merge sort needs at minimum two full passes over each worker's data: write sorted runs, then read and merge them. On top of that, the shuffle phase moves all records across the network once. For 10TB on a 100-node cluster, each node handles roughly 100GB of shuffled data. At 500MB/s sustained sequential SSD throughput, one read or write is about 3.3 minutes per node, so a full read-plus-write pass is about 6.7 minutes. Every extra merge pass adds another full read-plus-write pass. Maximize the initial in-memory sort chunk size and the k-way merge fan-in so all runs merge in a single pass.

30-Second Answer / Outline

1. Clarify input size, record format, sort key, output partition count, hardware, and whether the output must be globally sorted or only locally sorted.
2. Sample keys before the map phase and choose range boundaries that target equal record counts; use a hot-key strategy when one key dominates.
3. Map workers read sequential splits, sort memory-sized chunks, and write durable sorted runs partitioned by range.
4. Reduce workers perform a k-way merge for each range and publish each final partition through an atomic or conditional commit.
5. Persist task and phase state, heartbeat workers, retry failed tasks, and expose partition-boundary metadata with the output.

5-Minute Explanation

The single-machine primitive is external merge sort: read a memory-sized chunk, sort it, write a sorted run, and merge the runs with a min-heap. Distributed sorting adds a sample phase. The coordinator samples representative keys, selects N-1 range boundaries for N output partitions, and sends those boundaries to map workers.

Each map worker reads its input split sequentially, sorts chunks, routes records to the correct range, and flushes sorted runs to durable shared storage. The shuffle is the transfer of those runs to the reduce workers. Each reducer merges all runs for its range, writes to a task-attempt-specific temporary path, and atomically or conditionally publishes the final partition. Concatenating partitions in boundary order is globally sorted when ranges are disjoint.

The coordinator marks a task complete only after its durable artifacts are visible. If a worker disappears, a lease or heartbeat timeout reassigns the task; map runs already in shared storage survive, and a failed reduce task can repeat its merge without exposing partial final output. Skew, merge fan-in, storage buffers, and network contention determine whether the nominal linear scaling is achievable.

45-Minute Interview Approach

This is a discussion plan for the design question, not a claim that the article should take 45 minutes to read.

• 0-5 minutes — Clarify the contract: Confirm bounded versus streaming input, record format, sort-key semantics, output partition count, global ordering, duplicate-key ordering, and failure tolerance.
• 5-10 minutes — Estimate capacity: Relate input size to RAM, derive run counts, estimate sequential I/O time, shuffle volume, network time, and the likely bottleneck.
• 10-17 minutes — Establish the single-node primitive: Draw read → in-memory sort → sorted runs → k-way merge and explain why random I/O is avoided.
• 17-24 minutes — Add distributed partitioning: Introduce sampling, percentile boundaries, range routing, map workers, reducers, and the concatenation correctness argument.
• 24-30 minutes — Discuss skew: Compare uniform ranges, sample-based ranges, and virtual partitions/hot-key splitting; state how boundary metadata describes the result.
• 30-36 minutes — Optimize I/O: Calculate merge fan-in from RAM and buffer sizes, compare 2-way with k-way merging, and identify when a second merge level is unavoidable.
• 36-41 minutes — Add reliability: Cover shared run storage, durable task state, heartbeat timeouts, retry behavior, speculative attempts, atomic rename, and object-storage conditional publication.
• 41-45 minutes — Close with operations and trade-offs: Cover backpressure, cleanup, observability, partition metadata, input changes, and what streaming or custom-comparator extensions would require.

Core Entities

• SortJob - top-level job request: job ID, input and output paths, partition count, sort key, and overall status (PENDING, RUNNING, COMPLETED, FAILED).
• InputSplit - a byte range of the input file assigned to a specific map worker for reading.
• SortedRun - an intermediate sorted file written to shared storage by a map worker after sorting one in-memory chunk. Multiple runs are produced per worker and merged during the reduce phase.
• PartitionBoundary - a key value separating two adjacent output partitions. N output partitions require N-1 boundaries, derived from a sampled key distribution.
• OutputPartition - a fully sorted segment of the final output covering one key range, stored at a stable addressable path on distributed storage.
• WorkerTask - a unit of work (SAMPLE, MAP, or REDUCE) with a status, an assigned worker node, and a retry count. Schema details are in the deep dives.

API Design

Submit a sort job

POST /jobs
Content-Type: application/json

{
  "input_path": "s3://my-bucket/data/raw/",
  "output_path": "s3://my-bucket/data/sorted/",
  "sort_key": "$.event_time",
  "sort_order": "asc",
  "partition_count": 100
}

Returns 202 Accepted with a job_id. The job is asynchronous: the endpoint creates a SortJob record, enqueues the sample task, and returns immediately. Holding an HTTP connection open for a multi-hour sort is not a workable design.

Query job status

GET /jobs/{job_id}

Returns phase (SAMPLE, MAP, SHUFFLE, REDUCE, DONE) and per-phase task counts so the client can observe progress. The client polls until status: COMPLETED or status: FAILED.

Retrieve output locations

GET /jobs/{job_id}/artifacts

Returns the sorted partition files with their key ranges:

{
  "job_id": "job_abc123",
  "partitions": [
    { "partition_id": 0, "key_range": { "start": null, "end": "fence" }, "path": "s3://my-bucket/data/sorted/part-00000" },
    { "partition_id": 1, "key_range": { "start": "fence", "end": "jump" }, "path": "s3://my-bucket/data/sorted/part-00001" }
  ]
}

Key ranges let downstream consumers seek to the right partition without scanning every file. This endpoint is only meaningful after status: COMPLETED.

High-Level Design

The design builds in three steps: first, external merge sort on a single worker; next, horizontal scale via sampling and parallel shuffle; finally, fault tolerance via shared run storage and coordinator checkpointing. Each step adds exactly what the previous one can't handle.

Step 1: Sort data that doesn't fit in a single machine's RAM