Problem RestatementProblem
Design a key-value store that spreads data across many machines, like DynamoDB or Cassandra. Clients call put(key, value), get(key) and delete(key). The store must hold more data than one machine can, serve a very high request rate, and keep working when machines fail.
A common version gives numbers: "scale a single-node store to 50 million keys and 1 million requests per second, 50% reads and 50% writes". So we also walk through how to get from one node to a cluster.
RequirementsRequirements
1.1 Functional
put(key, value),get(key),delete(key).- Values up to about 1 MB.
- Configurable consistency per request (fast vs strongly consistent reads).
1.2 Non-Functional
- Scale out: add machines to get more capacity.
- High availability: keep serving when a machine or a whole rack fails.
- Durability: no acknowledged write is lost.
- Low latency: single-digit milliseconds.
1.3 Scale Estimates
- 1M requests/sec (500K reads + 500K writes).
- One well-tuned node handles about 50K–100K ops/sec, so we need ~20 nodes for throughput. With 3 copies of each write, writes triple, so plan for ~40–60 nodes.
- Data: 50M keys × 1 KB = 50 GB. That fits on one node, so this workload is limited by throughput, not storage.
1.4 API Design
/kv/{key}with the value, and header consistency: one|quorum|all/kv/{key}→ value + version/kv/{key}High-Level ArchitectureArchitecture
2.1 Overview
- Partitioning: split keys across nodes using consistent hashing. Picture a ring of hash values: each node owns a few slices of the ring, and a key belongs to the first node clockwise from
hash(key). - Virtual nodes: each physical machine owns many small slices (e.g., 256). Load then spreads evenly, and when a machine is added it takes a little data from everyone.
- Replication: each key is stored on N = 3 nodes (the owner plus the next 2 on the ring, placed in different racks or zones).
- Coordinator: whichever node receives the request forwards it to the replicas and waits for enough replies.
- Membership: nodes learn who is alive through a gossip protocol (each node regularly shares what it knows with a few random nodes).
2.2 Architecture Diagram
%%{init: {"look":"handDrawn","handDrawnSeed":7,"theme":"base","fontFamily":"Virgil, \"Segoe Print\", \"Comic Sans MS\", cursive","themeVariables":{"fontFamily":"Virgil, \"Segoe Print\", \"Comic Sans MS\", cursive","fontSize":"16px","primaryColor":"#fff4e6","primaryBorderColor":"#1e1e1e","primaryTextColor":"#1e1e1e","secondaryColor":"#e7f5ff","tertiaryColor":"#ebfbee","lineColor":"#1e1e1e","textColor":"#1e1e1e","mainBkg":"#fff4e6","nodeBorder":"#1e1e1e","clusterBkg":"#f8f9fa","edgeLabelBackground":"#ffffff","classText":"#1e1e1e"}}}%%
flowchart LR
C["Client library"] --> CO["Coordinator node"]
CO --> R1["Replica A - zone 1"]
CO --> R2["Replica B - zone 2"]
CO --> R3["Replica C - zone 3"]
R1 --- G["Gossip - membership and failure detection"]
R2 --- G
R3 --- G
R1 --> S1[("Commit log + LSM tree")]Storage Engine on Each Node
Most large KV stores use an LSM tree (Log-Structured Merge tree), which is great for heavy writes:
- Commit log: append the write to a file on disk first (so it survives a crash).
- Memtable: put it in a sorted in-memory table.
- When the memtable is full, write it to disk as an immutable sorted file (SSTable).
- Background compaction merges SSTables, dropping overwritten and deleted values.
- Reads check the memtable, then SSTables from newest to oldest. A Bloom filter per SSTable quickly says "this key is definitely not here", which skips most files.
Deletes write a tombstone (a "deleted" marker), so replicas that missed the delete don't bring the value back.
Consistency with Quorums
With N = 3 replicas, choose:
- W = how many replicas must confirm a write.
- R = how many replicas we read from.
If R + W > N (e.g., W=2, R=2), every read overlaps with at least one replica that has the latest write, which gives strong-ish consistency. If you want speed, use W=1, R=1, but reads may be stale for a moment (eventual consistency).
Conflicts: two clients may write the same key at the same time on different replicas. Options:- Last-write-wins using timestamps: simple, but may silently drop one write.
- Vector clocks (a version counter per replica): detect true conflicts and return both versions for the app to merge. This is more complex.
Handling Failures
- Hinted handoff: if replica B is down, another node stores the write temporarily with a note ("this belongs to B") and hands it over when B returns.
- Read repair: if a read sees replicas with different versions, it updates the stale ones in the background.
- Anti-entropy with Merkle trees: replicas compare a tree of hashes over key ranges. Only ranges with different hashes are synced, so repair is cheap even with huge data sets.
- Permanent failure: the node is removed, and its ring slices are re-replicated from the remaining copies.
Deep Dive — Climbing from 80K to 1M operations per secondDeep dive
The interviewer's number is a single node doing 80K ops/sec and a target of 1M, half reads and half writes. The answer is a sequence, and each step buys a different thing.
Make the one node faster
Add a cache in front, tune the storage engine, give it more memory, move to faster disks. Every one of these is worth doing and none of them changes the shape of the problem.
%%{init: {"look":"handDrawn","handDrawnSeed":7,"theme":"base","fontFamily":"Virgil, \"Segoe Print\", \"Comic Sans MS\", cursive","themeVariables":{"fontFamily":"Virgil, \"Segoe Print\", \"Comic Sans MS\", cursive","fontSize":"16px","primaryColor":"#fff4e6","primaryBorderColor":"#1e1e1e","primaryTextColor":"#1e1e1e","secondaryColor":"#e7f5ff","tertiaryColor":"#ebfbee","lineColor":"#1e1e1e","textColor":"#1e1e1e","mainBkg":"#fff4e6","nodeBorder":"#1e1e1e","clusterBkg":"#f8f9fa","edgeLabelBackground":"#ffffff","classText":"#1e1e1e"}}}%%
flowchart LR
C["1M ops/sec of clients"] --> N["One node - 80K ops/sec"]
N --> D[("Local disk")]
N --> X["920K ops/sec rejected or queued"]
N --> F["Node dies - the whole store is down"]The ceiling is one machine's CPU, memory and disk, and it is roughly a 12x shortfall. It is also a single point of failure, which the requirements ruled out separately.
Replicate for reads
Add followers and send reads to them. Reads now scale close to linearly with the number of replicas, and a follower can be promoted if the leader dies.
Half the workload is writes, and every write still goes through one leader. That caps the system at the leader's ~80K writes/sec no matter how many followers are added, which lands at about 160K ops/sec against a 1M target. Read replicas fix availability and read volume; they do not fix write volume.
Partition the keyspace, then replicate each partition
Split keys across partitions with consistent hashing, and give each partition its own leader and two followers. Writes for different keys land on different machines, so write throughput grows with the number of partitions.
%%{init: {"look":"handDrawn","handDrawnSeed":7,"theme":"base","fontFamily":"Virgil, \"Segoe Print\", \"Comic Sans MS\", cursive","themeVariables":{"fontFamily":"Virgil, \"Segoe Print\", \"Comic Sans MS\", cursive","fontSize":"16px","primaryColor":"#fff4e6","primaryBorderColor":"#1e1e1e","primaryTextColor":"#1e1e1e","secondaryColor":"#e7f5ff","tertiaryColor":"#ebfbee","lineColor":"#1e1e1e","textColor":"#1e1e1e","mainBkg":"#fff4e6","nodeBorder":"#1e1e1e","clusterBkg":"#f8f9fa","edgeLabelBackground":"#ffffff","classText":"#1e1e1e"}}}%%
flowchart LR
SC["Smart client - knows the ring"] -->|"hash(key)"| P1["Partition 1 - leader + 2 replicas"]
SC --> P2["Partition 2 - leader + 2 replicas"]
SC --> P20["Partition 20 - leader + 2 replicas"]
P1 --> R1[("50K keys")]
P2 --> R2[("50K keys")]
P20 --> R20[("50K keys")]Twenty partitions at 80K ops/sec each clears the target with headroom. Two details make it real:
- Smart clients hold the ring and route straight to the owning node, saving a hop per request. The alternative, a coordinator tier, adds latency and another thing to scale.
- Virtual nodes let a new machine take small slices from many peers instead of splitting one neighbour, so rebalancing moves data gradually and can be throttled under live traffic.
Trade-offs & AlternativesTrade-offs
| Decision | Choice | Why | Alternative |
|---|---|---|---|
| Partitioning | Consistent hashing + virtual nodes | Small data movement when scaling | Range partitioning: good for range scans, needs splitting of hot ranges |
| Replication | Leaderless with quorums | High availability, no failover step | Leader per partition (Raft): simpler strong consistency, failover pause |
| Conflicts | Last-write-wins + conditional writes | Simple for most apps | Vector clocks/CRDTs: no lost writes, complex |
| Storage | LSM tree | Fast writes | B-tree: faster reads, slower random writes |
Common Follow-up QuestionsFollow-ups
- "What does a synchronously replicated hash map look like?" The writer sends each write to the replica and only confirms after the replica acknowledges. If the replica is down, the write fails, or the system switches to a new replica after a membership change. This trades availability for zero data loss.
- "How do you support range queries?" Use range partitioning (sorted keys, as in Bigtable or HBase) instead of hashing.
- "Multi-region?" Replicate asynchronously between regions, and use last-write-wins or route writes for each key to a home region.
Wrap-UpWrap-up
Spread keys with consistent hashing and virtual nodes, keep 3 replicas in different zones, and tune consistency with R and W quorums. Store data in an LSM tree with a commit log for fast, durable writes. Heal failures with hinted handoff, read repair and Merkle-tree sync, and handle hot keys with caching or key splitting.