Design a distributed key-value store
Build the storage layer itself: a store that keeps key → value pairs across hundreds of machines, keeps working when machines and whole zones fail, and lets each application choose how consistent it needs to be. This is the design behind Amazon's Dynamo paper and, in spirit, Cassandra, ScyllaDB and Riak. It ties together almost all of Level 3, and it follows the interview framework phase by phase.
1. The prompt
"Design a distributed key-value store like DynamoDB or Cassandra. Clients put and get values by key; the store must scale out by adding machines and keep working when machines fail."
There are no product features to discuss, so the interview is about mechanisms: how keys are spread over machines, how copies are kept, what "consistent" means when copies disagree, and how a cluster of hundreds of nodes heals itself every week without anyone touching it.
2. Frame the problem
What operations? Just get and put, or ranges and queries too?
Get, put and delete by key. No range scans, secondary indexes or multi-key transactions.
Pure key lookups allow hash partitioning, which spreads load evenly but rules out ordered scans.
How big are values, and how much data in total?
Values up to about 1 MB, mostly a few KB; about 500 TB of data.
Storage, not throughput, will set the node count.
When nodes or a zone fail, should writes keep working?
Yes. Being always writable matters more than every read seeing the latest write.
This is the availability-over-consistency (AP) choice that shapes everything below.
Can each request choose its consistency?
Yes: a fast, possibly stale read, or a quorum read that sees the latest acknowledged write.
We need tunable N, W and R per request.
What if two clients write the same key at the same time?
For most data the last write can win; some applications want to keep both and merge.
Versioning: timestamps for simplicity, vector clocks or mergeable types where it matters.
How fast?
p99 under 10 ms within a region, at about 2M operations a second.
One network round trip to a few replicas, and a storage engine that writes sequentially.
One region or several?
One region with three zones to start; multi-region later.
Replica placement must spread copies across zones.
3. What it must do
- A client can put a value under a key, choosing how many copies must confirm.
- A client can get the value for a key, choosing how many copies to consult.
- A client can delete a key.
Not now:
- Range queries and secondary indexes: they need ordered partitioning and a different design.
- Multi-key transactions: they need consensus per range (the CP alternative in section 12).
- A query language: belongs in a layer above.
4. How well it must do it
| Target | Why this number |
|---|---|
| Always writable: writes succeed with any one node down, and with a whole zone down consistency or availability | For a shopping cart or session store, rejecting a write is worse than reconciling later. |
| Fast: p99 under 10 ms for reads and writes within a region speed | Applications call the store several times per page. |
| Scale out: 500 TB and 2M operations a second, growing by adding nodes with no downtime load shape | Capacity grows constantly; big-bang migrations aren't acceptable. |
| Durable: an acknowledged write survives node and zone failures data safety | It's the system of record for the applications on top. |
| Self-healing: routine disk and node failures need no human surviving failures | At hundreds of nodes, something fails every week. |
5. Numbers that change the design
| Quantity | Working | → so we… |
|---|---|---|
| Data | 500 TB × 3 replicas | 1.5 PB raw; at ~4 TB usable SSD per node, ~400 nodes → so storage sets the size of the cluster |
| Throughput | 2M operations/s; one node handles ~20–50K | ~50–100 nodes' worth → so throughput is comfortably covered by the storage-sized cluster |
| Failures | 400 disks, a few percent failing a year | A disk dies every week or two → so repair must be automatic, normal operation |
| Adding a node | Hash-mod-N with N = 400 → 401 | Almost every key would move → so we need consistent hashing |
6. The nouns
- Key and value.
- Version: a timestamp or a vector clock.
- Node.
- Token range: the part of the hash ring a node owns.
- Replica set: the N nodes holding one key.
- Hint: a write held for a node that was down.
- Tombstone: a delete marker.
on each node LSM tree: commit log → memtable → SSTables (+ Bloom filters), compaction per key key → [(value, version)] (several versions only if writes were concurrent) cluster state ring: token → node, node → zone, node → status (spread by gossip)
7. The contract
A client library (or any node, acting as coordinator) exposes three calls. The consistency level and a version context travel with them.
put(key, value, consistency = QUORUM, context?) → ok
get(key, consistency = QUORUM) → [value(s)], context // more than one value = concurrent versions
delete(key) → ok // writes a tombstone
The context carries version information (for example a vector clock), so the client can say which version its write is based on. Consistency levels: ONE (fastest), QUORUM (a majority of the N replicas), ALL.
8. First working version
Start with the simplest store that spreads data: a client library that hashes each key to one of the nodes, and nodes that store values in a local engine.
8.1 Putting a value
The client computes hash(key) mod N to pick the node and sends the write there. That node appends it to its commit log, updates its in-memory table, and acknowledges.
8.2 Getting a value
The same hash finds the same node, which reads its memtable, then its on-disk files (using Bloom filters to skip files that can't contain the key), and returns the value.
8.3 Deleting a key
The node removes the key from its engine. With one copy, that's all.
Parked for later (section 9 fixes each)
- Adding a node changes
mod N, so almost every key moves. - One copy per key: a dead disk loses data, and a down node makes keys unavailable.
- With copies, which one wins when two clients write the same key at once?
- Copies drift apart after failures; how do we find and fix differences cheaply?
- Who knows which nodes are alive, without a single coordinator?
- With copies, a plain delete can come back from the dead.
9. Make it hold up
9.1 Spreading keys so nodes can come and go scale out
| Approach | Why | |
|---|---|---|
| ✗ Bad | hash(key) mod N | Changing N remaps nearly every key; growing the cluster means moving almost all the data. |
| ~ Good | A consistent hashing ring: each node owns the arc up to its position; a key goes to the next node clockwise | Adding a node moves only the keys in one arc. But one position per node gives uneven arcs. |
| ✓ Great | A ring with virtual nodes (many tokens per physical node), with bigger machines owning more tokens | Load spreads evenly, and a new node takes small slices from many others, so rebalancing is fast and gentle. Module 15 has an interactive ring. |
9.2 Copies and quorums always writable · durable · fast
- Replication: each key lives on the next N = 3 distinct nodes clockwise, chosen in 3 different zones, so a zone failure leaves two copies.
- Any node coordinates: a client sends a request to any node (or computes the replicas itself); the coordinator forwards it to the N replicas.
- Quorums: a write is acknowledged after W replicas confirm; a read waits for R replies. If W + R > N (say 2 + 2 > 3), every read overlaps every acknowledged write on at least one replica, so it sees the latest version (Module 14).
- Hinted handoff: if a replica is down, the coordinator still needs W confirmations, so another healthy node stores the write with a hint ("this belongs to node C") and replays it when C returns. This "sloppy quorum" keeps the store writable; the price is that a read right afterwards may miss the write unless R + W > N on the proper replicas.
9.3 Concurrent writes to the same key always writable · durable
| Approach | Why | |
|---|---|---|
| ✗ Bad | Last write wins by each node's own clock | Clock skew (Module 20) picks the wrong winner, and concurrent updates silently vanish. |
| ~ Good | Last write wins with timestamps from the client or coordinator, kept well synchronised | Simple and widely used (Cassandra does this); fine when values are overwritten whole and losing a concurrent update is acceptable. |
| ✓ Great (when it matters) | Vector clocks: return all concurrent versions ("siblings") on read and let the application merge them, or use CRDT values (counters, sets) that merge automatically | No update is lost silently; the original Dynamo merged shopping carts this way. It costs complexity for clients. |
9.4 Finding and fixing divergence cheaply self-healing · durable
| Approach | Why | |
|---|---|---|
| ✗ Bad | Periodically compare every key between replicas | Moves terabytes over the network to find a handful of differences. |
| ~ Good | Read repair only: fix stale replicas when a read notices them | Free, but keys that are rarely read stay inconsistent indefinitely. |
| ✓ Great | Read repair plus Merkle trees per token range: each replica hashes its keys into a tree; replicas compare roots, then only the branches that differ | Finds the differing keys by exchanging a few hashes. Try it in section 13. |
9.5 Membership and failure detection without a coordinator self-healing
There's no central coordinator to fail. Each node periodically gossips its view of the cluster (who's alive, which tokens they own) with a few random peers; the information reaches every node in a few rounds. An accrual failure detector marks a node "suspected down" by how late its heartbeats are, instead of a fixed timeout. A node marked down keeps its tokens (no data moves); replicas and hints cover for it. Data moves only for planned changes: adding or removing capacity.
9.6 Deletes, and the storage engine durable · fast
- Tombstones: a delete can't simply remove the value, because a replica that missed it would bring it back during repair. Write a tombstone (a delete marker with a version) that's replicated like a write and removed by compaction only after a grace period long enough for every replica to have seen it.
- An LSM engine on each node (Module 12): writes go to a commit log and an in-memory table, flushed as sorted immutable files and merged by compaction. Writes are sequential, so they're fast; Bloom filters keep reads from opening files that can't hold the key.
- The cost: compaction uses disk bandwidth in the background and must be throttled, and many tombstones slow down reads until they're compacted away.
9.7 The whole design after the deep dives
A write while one replica is down, then a read that repairs the stale copy, through the final design.
10. Interviewer follow-ups
A customer needs strongly consistent reads and compare-and-set.
Quorum reads and writes give read-your-writes for acknowledged writes, but not linearizable compare-and-set under concurrency. For that, run a consensus group (Raft or Paxos) per token range, as TiKV and CockroachDB do, or offer a lightweight-transaction mode that uses Paxos for those operations only, as Cassandra does. It is slower and gives up availability when a range has no majority.
One key is read 100,000 times a second.
Consistent hashing balances keys, not traffic, so that key's three replicas are overwhelmed. Let clients cache it briefly, read from any replica with consistency ONE, or split the key into several sub-keys and combine (Module 15).
Add 50 nodes to a 400-node cluster without downtime.
Each new node takes tokens from many existing nodes; data for those ranges streams to it in the background, throttled, while the old owners keep serving. Once streaming completes, ownership flips in the ring via gossip. Nothing is unavailable at any point.
Expand to a second region.
Replicate each key to N replicas in each region. Clients use LOCAL_QUORUM (a majority within their own region) for low latency, accepting that the other region may be a little behind. Cross-region conflicts use the same versioning rules.
A node is down for a week.
Hints have a time limit (say three hours), so after that they're dropped. When the node returns, anti-entropy with Merkle trees finds and streams the differences. If the node won't return, replace it: a new node takes its tokens and streams the data from the other replicas.
Why not let a single leader handle each key?
Leader-based replication gives simpler consistency, but writes stop for that range while a new leader is elected, and the leader is a hot spot. The requirement was "always writable", which leaderless quorums give. It's a trade-off worth stating.
How do clients find the right nodes?
Either any node acts as coordinator and forwards requests (one extra hop), or a "smart" client library learns the ring from the cluster and sends requests straight to the replicas (fewer hops, more complex clients). Most production drivers do the latter.
Large values, say 50 MB?
Keep the store for small values. Put large blobs in object storage and store their key in the KV store. Huge values hurt compaction, repair and tail latency for everyone.
11. When things break
| Failure | What users see | How we detect it | What limits the damage |
|---|---|---|---|
| One node down | Nothing (W and R still reachable) | Gossip and the failure detector | Hinted handoff; reads use the other replicas |
| A zone down | Slightly higher latency | A third of nodes suspected | Replicas are in three zones, so two copies remain; quorums still met |
| Disk failure | Nothing | Node errors; node marked down | Replace the node; it streams data from its replicas |
| Replicas diverge (missed writes) | Possibly a stale read with consistency ONE | Read repair notices; Merkle comparisons | Read repair and scheduled anti-entropy |
| Compaction falls behind | Slower reads, growing disk use | Pending compactions, read latency | Throttle writes, add nodes, tune compaction |
| Clock skew with last-write-wins | A newer update lost | Hard to see; audits | Tight NTP, coordinator timestamps, or vector clocks for sensitive data |
12. Wrap-up and the clock
Trade-offs we made:
- Availability over consistency (AP), with tunable quorums per request.
- Leaderless replication with hinted handoff instead of leaders per range.
- Last-write-wins by default, with siblings or CRDTs where data matters.
- LSM storage (fast writes, compaction cost).
The CP alternative is a consensus group per range: linearizable, but a range without a majority stops accepting writes.
What we'd watch:
- p99 latency per consistency level.
- Hints pending.
- Read-repair rate.
- Pending compactions.
- Disk use per node.
- Nodes suspected down.
At 10×:
- More nodes and more tokens per node.
- Multi-region with LOCAL_QUORUM.
- Smart clients.
A 45-minute interview
| 0–4 | Frame the problem (operations, AP vs CP, size) |
| 4–9 | Must do, how well, numbers |
| 9–12 | Nouns and contract |
| 12–19 | First working version, parked list |
| 19–40 | Deep dives: ring (9.1), quorums and hints (9.2), conflicts (9.3), repair (9.4) |
| 40–45 | Follow-ups and wrap-up |
A 60-minute interview
| 0–5 | Frame the problem |
| 5–11 | Must do, how well, numbers |
| 11–15 | Nouns and contract |
| 15–23 | First working version |
| 23–50 | Deep dives 9.1–9.6 and the final picture |
| 50–60 | Follow-ups: strong consistency, hot keys, adding nodes; wrap-up |
| Level | What a strong answer shows |
|---|---|
| Mid-level | Hash partitioning with consistent hashing, replication factor 3, quorum reads and writes, a basic storage choice. |
| Senior | W + R > N explained, hinted handoff, read repair, tombstones, LWW vs vector clocks with their costs, zone-aware replica placement. |
| Staff+ | Merkle-tree anti-entropy, gossip and accrual failure detection, rebalancing without downtime, AP vs CP per range, operations (compaction, repairs, capacity). |
Common mistakes:
hash mod N.- Replicas in one zone.
- Last-write-wins with node clocks.
- Deletes without tombstones.
- Moving data whenever a node is briefly down.
- Claiming strong consistency without consensus.
13. Try it, check yourself, read more
Two replicas each hold 8 keys, hashed into a Merkle tree. Click keys to make replica B's copy differ, then compare: matching hashes (green) are skipped entirely; only branches with a difference (red) are explored.
What is hinted handoff?
Why are deletes written as tombstones?
With Merkle trees, what do two replicas exchange to find differences?
Why does W + R > N matter?
| Resource | Why |
|---|---|
| Dynamo: Amazon's highly available key-value store | The paper this design follows, section by section. |
| Cassandra's Dynamo architecture | How a production system implements the same ideas today. |
| Designing Data-Intensive Applications, chapters 5–6 | Leaderless replication and partitioning in depth. |