Design a Distributed Key-Value Store
Problem
Design a key-value store that exposes get(key) and put(key, value) and runs on a cluster of commodity machines. It must store 100 TB of data, with values up to 1 MB and a typical size of 1 KB. The cluster serves 1 million reads and 500,000 writes per second at peak, and it must stay available for reads and writes when individual nodes fail or the network partitions. Operators can add or remove nodes without downtime, and clients can choose, per request, how strongly consistent a read must be.
- Do we favor availability or consistency during a network partition?
- Is eventual consistency acceptable by default, with an option for stronger reads?
- Do we need range scans over keys, or only point lookups?
- Do we need multi-key transactions?
- Is the cluster in one data center, or replicated across regions?
- How long must data survive: is it a cache or the system of record?
Items in bold are existing requirements. The rest are suggestions. Confirm with the interviewer whether they are relevant first.
put(key, value)stores a value, andget(key)returns the latest value.delete(key)removes a key.- Clients choose a consistency level per request (for example one replica, or a quorum).
- Add and remove nodes online, with data rebalanced automatically.
- Support an optional time-to-live on keys.
- Detect and resolve conflicting writes made during a partition.
Items in bold are existing requirements. The rest are suggestions. Confirm with the interviewer whether they are relevant first.
- Store 100 TB of logical data; with three replicas that is 300 TB, or about 150 nodes with 2 TB of usable disk each.
- Serve 1 million reads and 500,000 writes per second, which is about 7,000 reads and 3,500 writes per node.
- Remain available for reads and writes despite node failures and network partitions.
- p99 latency under 10 ms for single-key operations within a data center.
- Durability: an acknowledged write survives the loss of any single node.
- Linear scalability: doubling nodes roughly doubles capacity and throughput.
Keys are spread across nodes with consistent hashing and virtual nodes, and each key is replicated to the next N distinct nodes on the ring. Any node can coordinate a request and uses tunable quorums (W replicas must acknowledge a write, R must answer a read) to trade consistency against latency. Nodes discover membership and failures through gossip, repair missed writes with hinted handoff and anti-entropy, and store data locally in a log-structured merge tree for fast writes.
Loading diagram...
Write. The client library hashes the key and sends put to one of the key's replicas, which acts as coordinator. The coordinator attaches a version (a hybrid logical timestamp, or a vector clock in the conflict-preserving mode) and sends the write to all N = 3 replicas in parallel. Each replica appends to its commit log, updates its in-memory memtable, and acknowledges. The coordinator returns success once W = 2 replicas acknowledge. If a replica is down, the coordinator writes the update to a healthy stand-in node as a hint, which is replayed when the replica returns.
Read. The coordinator asks R = 2 replicas for the key (a full value from one, digests from the others). If the versions agree, it returns the value. If they differ, it returns the newest version (or all conflicting siblings) and asynchronously writes the newest version back to the stale replica, which is called read repair. Because R + W > N, every quorum read overlaps the latest quorum write.
Membership change. A new node joins by gossiping its token ranges. Neighboring nodes stream the affected key ranges to it, and once streaming completes, the ring metadata is updated and the new node starts serving those ranges.
Maps each key to the nodes that own it.
- Consistent hashing with a 64-bit hash (for example Murmur3) and about 256 virtual nodes per physical node.
Diagram
Loading diagram...
Scalability
Adding a node moves only about 1/N of the keys, taken in small slices from many existing nodes, so no single node is overwhelmed during rebalancing. Virtual nodes also let a bigger machine own proportionally more ranges. Replica placement skips vnodes on the same physical node and, when configured, the same rack.
Tradeoffs
Simple modulo hashing (hash(key) mod nodes) reshuffles almost every key when the cluster size changes. Range partitioning would support ordered scans but creates hot spots for sequential keys and needs a separate balancer. Consistent hashing gives even spread and cheap membership changes, which matter more for a point-lookup store.
Keeps N copies of each key and coordinates reads and writes across them.
API
put(key, value, { consistency: ONE | QUORUM | ALL, ttl? })get(key, { consistency }) -> { value, version } or siblingsdelete(key)writes a tombstone with a version.
Scalability
Every node can coordinate, so there is no master bottleneck, and a request needs only W or R replica responses, so one slow node does not set the latency. Requests go to replicas in the local data center first, with cross-region replication done asynchronously.
Tradeoffs
Leaderless quorum replication keeps writes available during partitions, which is the stated priority, at the cost of possible conflicts. A single-leader design per partition (as in Raft-based stores) gives linearizable reads but loses write availability while a leader is elected. Defaulting to last-write-wins with hybrid clocks is simple and fine for most data. Vector clocks with sibling values are offered for data such as shopping carts where losing a concurrent write is unacceptable and the client can merge.
Tracks which nodes are alive and which token ranges each owns.
- Gossip protocol (each node exchanges state with a few random peers every second) with a phi accrual failure detector.
- Seed nodes for bootstrapping.
Data Model
- Per node:
{ nodeId, address, tokens[], status, heartbeatVersion, generation }, merged by taking the higher heartbeat version.
Scalability
Gossip spreads a change to all 150 nodes in a few rounds (logarithmic in cluster size), and each node sends only a few small messages per second, so no central registry is needed.
Tradeoffs
A central coordinator such as ZooKeeper would give an authoritative membership view but adds an external dependency and a potential bottleneck. Gossip is decentralized and robust, but nodes can briefly disagree about membership, which quorums and hinted handoff tolerate. Permanent removal of a node is an explicit operator action, so a temporary outage never triggers a costly full rebalance.
Background process that makes replicas converge after missed writes.
- Merkle trees per token range, compared between replicas.
Scalability
Replicas exchange tree roots first and descend only into subtrees whose hashes differ, so repairing a range transfers data proportional to the differences, not the range size. Repairs are rate limited and scheduled per range to avoid saturating disks.
Tradeoffs
Hinted handoff and read repair fix most divergence cheaply but miss keys that are never read and hints that expire. Periodic Merkle tree repair closes that gap. Tombstones must be kept longer than the repair interval (for example 10 days), or a deleted key can be resurrected from a replica that missed the delete.
Per-node storage that makes writes sequential and reads efficient.
- Commit log, in-memory memtable (a skip list), and immutable SSTables on SSD, as in RocksDB.
- Bloom filters and a sparse index per SSTable.
Data Model
- SSTable entries:
key,version,value or tombstone,ttl, sorted by key.
Scalability
Writes are an append to the log plus an in-memory insert, so each node handles thousands of writes per second. Reads check the memtable, then use Bloom filters to skip SSTables that cannot contain the key. Compaction merges SSTables in the background, dropping overwritten values and expired tombstones.
Tradeoffs
A B-tree engine has more predictable reads but turns writes into random disk I/O. The LSM tree suits a write-heavy store, at the cost of compaction overhead and some read amplification. Leveled compaction is chosen for read-heavy tables (less amplification) and size-tiered compaction for write-heavy ones.