Key-Value Store

What the interviewer is asking

"Design a key-value store" (sometimes "design Dynamo") asks for a huge hash map spread over many machines: put a value under a key, get it back by that key. Shopping carts, sessions and user settings often live in one.

Storing one key is easy. The question is what happens at scale: which machine holds each key, where the copies go, and what a get returns when machines die or the network splits and the copies disagree. Amazon's Dynamo paper (2007) described this kind of design, and stores such as Cassandra and Riak are built along similar lines. The walkthrough follows the 8 steps of the interview framework.

1. Scope: what version 1 must do

Version 1 is three calls by key: put, get and delete. Then the promises that shape the design: values up to about 1 MB, highly available (puts and gets always work), room to grow to many machines, and still working when a machine, or a whole rack of machines, fails.

Park the rest. Range queries would ask every machine, because spreading keys by hash scatters neighbors. Changing two keys at once, all or nothing, needs a protocol across machines. Finding keys by a field inside their values needs a secondary index, and the store never looks inside a value.

2. Data: a value and a version on every copy

To survive a dead machine, every key is kept on 3 machines. Right after a put, a copy that was slow or down can lag behind, so the copies briefly disagree. Each copy therefore stores key → value + version: the value is just bytes, and the version grows with every put.

copy     value        version
copy 1   milk, book   8
copy 2   milk         7
copy 3   milk         7

A majority vote would return the old cart: two copies say "milk" only because the newest put hasn't reached them yet. The version says copy 1 is newest. Inside one machine the store is a hash map plus a disk; the Time Based Key-Value Store coding lesson builds a small single-machine version.

3. The API: three calls by key

PUT    /keys/ana   {"value": "milk, book, pen"}   → 200 OK
GET    /keys/ana    → 200 OK  {"value": "milk, book", "version": 8}
DELETE /keys/ana    → 204 No Content
GET    /keys/bob    → 404 Not Found   (never stored)

Clients name their own keys, so storing a value is a PUT to the key's address, and a PUT sent twice leaves the same result: a client can safely retry one that timed out. POST is for when the server picks the address, and a GET must never change data. If the copies disagree, the reply carries the newest version, or, with the conflict handling in section 9, two versions for the client to merge.

4. The busy path: get by key

Clients make about 10 gets for every put, so the get is the path to keep fast. On one machine it's short: the app sends GET /keys/ana, the store finds ana in its index of keys, reads the value and its version from disk, and answers.

5. The numbers

From the interviewer: 10 TB of keys and values, 3 copies of every key, and 200,000 requests a second at peak. Say one machine has about 1 TB of disk to use.

• Storage: 10 TB × 3 copies = 30 TB, so at least 30 machines. Forgetting the copies gives 10.
• Headroom: data grows, and a dead machine's data must be copied onto the others, so keep disks about 3/4 full: about 40 machines.
• Per machine: 200,000 ÷ 40 = 5,000 requests a second each. One machine taking every request would do the work of all 40.
• Per copy: once each get reads 2 copies and each put writes 3 (sections 7 and 8), the machines do about 420,000 reads and writes a second, about 10,000 each. Keep each machine's hottest keys in memory.

6. Services: a ring, and no machine in charge

With hash(key) mod 40, a 41st machine would send about 40 in 41 keys somewhere else, and nearly all 30 TB would be copied over the network. Use consistent hashing instead: machines and keys hash onto one ring, and a key lives on the first machine clockwise. A new machine takes only the keys just before it, its share of about 1 in 41. With virtual nodes (each machine at many points) the slices come out even. A small ring:

positions   0 to 99, clockwise from the top
machines    A 8, B 25, C 42, D 58, E 75, F 92
keys        ana 16 → B, zoe 84 → F

No machine is in charge. Any machine can take a request: it becomes that request's coordinator, finds the key's machines on the ring, forwards the request and passes the answer back. A leader would take all 200,000 requests a second, and its death would stop the store. Clients can also keep a copy of the ring and go straight to the right machine. To keep the ring current, machines gossip: every second each one swaps its list of machines, and which are alive, with a random other one, so news of a new or dead machine spreads within seconds.

7. Scale: 3 copies on the ring, on 3 racks

A key kept only on B dies with B. The copies go on the owner, then the next machines clockwise: ana lives on B, C and D. With virtual nodes the walk skips points of a machine that already has a copy. If C leaves for good, the first 3 machines from ana are B, D and E, so E copies ana from B or D. If C is only down for a few minutes, another machine can keep C's new writes and hand them back when it returns: hinted handoff.

Machines sit in racks that share power and a network switch, so a whole rack can fail at once. If 2 of a key's copies share a rack, that one failure leaves a single copy. So the walk also skips a machine whose rack already has a copy:

machine   A  B  C  D  E  F
rack      1  2  3  1  3  2

zoe's owner is F (rack 2) and A, next, is on rack 1. B shares rack 2 with F, so skip it; C, on rack 3, takes the third copy. Any one rack can now fail and zoe keeps 2 copies.

8. Deep dive: quorums

Recall the overlap rule from CAP and consistency: a put goes to all N copies but waits for W acks, and a get asks R copies and keeps the newest version. When R + W > N, the copies written and read must share one, so no get misses a confirmed put. N = 3, W = 2, R = 2 is a quorum.

Say D is down and a put writes version 9: B and C ack, so it's confirmed. D comes back still holding version 8, and a get asks any 2 copies:

copies asked   versions   returns
B and C        9, 9       9
B and D        9, 8       9
C and D        9, 8       9

With one machine down, puts and gets both work and stay correct. With C and D both down, W = 2 can't be met. Refusing the put is the consistent answer, right for money or usernames. This store promised availability, so it accepts the put on B alone (W = 1) and copies it to C and D later. The price: a get reading only C and D misses it (1 + 2 = 3, not more than 3), and two sides of a split can accept different puts for one key. Many Dynamo-style stores let each request choose its own W and R.

9. Deep dive: conflicts

Say the network splits: B can't reach C or D, and both sides keep taking puts. Ana's cart holds milk, written at time 100. Her laptop, on B's side, adds a book (time 205); her phone, on the other side, adds a pen (time 210).

With last-write-wins, every copy keeps the latest value once the network heals: milk, pen. The book is silently lost, and if B's clock ran fast, the older put could win instead. That's fine where losing a put is harmless, like a profile photo changed twice; Cassandra settles conflicting writes by timestamp this way.

A cart must never lose an item, so keep both versions: a get returns both, the cart code merges them into milk, book, pen, and writes the merge back. To tell "one replaced the other" from "written side by side", each version carries a vector clock, one counter per machine that wrote it:

value        vector clock   written via
milk         B: 1           B
milk, book   B: 2           B (the laptop)
milk, pen    B: 1, D: 1     D (the phone)

"milk, book" is ahead on B's counter but behind on D's, so neither version saw the other: both are kept. Riak can keep both versions this way, and the Dynamo paper described it for its shopping cart, including the catch: a merge can bring back an item removed on one side.

10. Deep dive: repair

D missed version 9 while it was down. When a get asks B and D, it hears 9 and 8, answers version 9, and sends version 9 to D on the spot: read repair. Doing nothing would leave ana with only 2 up-to-date copies until the next put.

Keys nobody reads would stay old, so a background job also compares machines: anti-entropy. Each machine keeps a Merkle tree, a tree of hashes where each leaf hashes a range of keys and each parent hashes its children. Two machines compare roots, walk down only the branches that differ, and send just those ranges.

What breaks first

• Deletes: simply removing a key lets a copy that missed the delete bring it back during repair. Write a marker with a version instead (a tombstone), and clean it up once every copy has it.
• A hot key: all its gets land on the same 3 machines. Cache it in memory in front of the store.
• Values over 1 MB: keep them in object storage and store only a pointer under the key.

Summary: the order to follow

1. Scope: put, get, delete; values up to 1 MB; always available, survives a machine or a rack failing.
2. Data: key → value + version, on every copy.
3. API: PUT, GET and DELETE /keys/{k}.
4. Busy path: get by key, about 10 gets for every put.
5. Numbers: 30 TB with copies, about 40 machines, 5,000 requests a second each.
6. Services: keys on a consistent hashing ring; any machine coordinates.
7. Scale: each key on its owner and the next 2 machines clockwise, on 3 racks.
8. Deep dives: quorums (W = 2, R = 2; W = 1 when 2 copies are down), last-write-wins or merged versions, read repair and anti-entropy.

The same steps work for any storage question: where each key lives, how many copies must answer, and what to do when they disagree.