Loading…

Understanding the problem

A key-value store keeps a value under a key and gives it back. Here it is the durable store many teams share, about 30 TB of carts, sessions and usernames, so a lost write is lost data. Caching in front of it is a different problem.

Three things make it hard: spreading keys so machines can come and go, keeping copies that survive a zone while staying fast, and being honest about what a read sees. Leaderless copies, as in Dynamo and Cassandra, are one family of answer; consensus per key range, as in Raft-based stores, is the other. This page leads with the first and credits the second throughout.

The practice brief gives the requirements; confirm them quickly. The first non-functional one is the invariant: at the default setting, an acknowledged write survives the loss of any one machine or zone.

Functional requirements

CoreTeams get, put (with an optional expiry) and delete single keys in their own namespaces.
CoreA team chooses, per request, between the fastest answer and one that includes every acknowledged write.
CoreA team can write a key only if it doesn’t exist yet, or only if it hasn’t changed since it was read.
CoreEach namespace has its own settings and limits.
Below the lineRange scans and secondary indexes.
Below the lineTransactions across several keys.
Below the lineValues over 1 MB (kept in object storage).
Below the lineA query language.
Below the lineA cache in front of the store.That is its own problem; here every read comes from the store’s copies.
Below the lineA stream of changes for other systems.Useful, and built on the same log, but not what keeps writes safe.

Non-functional requirements

DurabilityAt the default setting, an acknowledged write survives the loss of any one machine or any one zone.
AvailabilityReads and writes keep working while one machine or one zone is down.
ConsistencyA client that asks for it reads its own latest acknowledged write; weaker reads are allowed when a team chooses them.
ConsistencyWhen two clients write one key at the same moment, the result is predictable, and the design says what is kept and what can be lost.
LatencyA single-key get or put takes under 10 ms at p99 within a region, at the default setting.
Quality

Scale

From the briefNumbers here are given in the practice brief.

ScaleAbout 30 TB of live data before copies, about 30 billion keys, growing about 50% a year; values about 1 KB typically and up to 1 MB.
ScaleAbout 500,000 reads and 150,000 writes a second at peak; one team’s keys can take 10 times their usual writes during a sale.
ScaleMachines have SSDs of about 8 TB; they restart for patches weekly, and a disk or machine dies about once a week.
ScaleOne region with three zones today (a zone lost power for 40 minutes once in two years); a second region, about 80 ms away, within a year.

Out of scope

Out of scopeThe teams’ own services and what they store.
Out of scopeAuthentication internals: services already have identities.
Out of scopeClient libraries for every language.
Out of scopeHardware purchasing and billing.

Check yourself

Why is “the store says saved” the moment that matters?
Show answerHide answer
After it, the client moves on, so the write must already be on enough machines, in enough zones, to survive a failure. Copying to the third zone can happen later.

The numbers that shape the design

From the briefNumbers here are given in the practice brief.

Work out the few numbers this design depends on: the data per machine, not the request rate, sizes it.

Back of the envelope from the brief

Quantity	Value	Working
Data with copies	About 90 TB	30 TB of live data × 3 copies
Machines	About 30	90 TB ÷ about 3 TB each, leaving 8 TB disks mostly free
Keys	About 30 billion	30 TB ÷ about 1 KB
Writes per machine	About 15 MB/s	150,000 writes/s × 1 KB × 3 copies ÷ 30
Reads per machine	About 17,000/s	500,000 reads/s ÷ 30, if each asks one copy
Rebuilding one machine	About 8 hours	3 TB at 100 MB/s from a single peer

What the numbers tell you

The request rate is modest for 30 machines.
Rebuilding one machine from a single peer takes most of a working day.
The data grows by half each year, so machines are added every few months.

Architecture diagramWhere the load lands

From the brief: the data on each machine and the hours to rebuild one shape the design, not the request rate.

Diagram as text

Components: 30 TB live, 90 TB with copies (NoSQL DB); ~30 machines, ~3 TB each (Service); ~15 MB/s written per machine (Service); Rebuild one: ~8 h from one peer (Service). Connections: 30 TB live, 90 TB with copies → ~30 machines, ~3 TB each (about 3 TB per machine); ~30 machines, ~3 TB each → ~15 MB/s written per machine (150,000 writes/s); ~30 machines, ~3 TB each → Rebuild one: ~8 h from one peer (a machine dies).

Read:Learn: Numbers and capacity estimation

Try it: The numbers that shape the designGuided practice · Distributed key-value store, Shared key-value store · Senior

Keep reading+1 min

A year of growth

At 50% a year the data reaches about 135 TB with copies, about 45 machines. Adding them is a planned, throttled move, so it belongs before the season’s peak, not during it.

Check yourself

Why keep only about 3 TB on an 8 TB disk?
Show answerHide answer
The machine needs room to merge its files and to take over a dead machine’s data; a full disk stops it taking writes.

High-level design

Build the design in three steps: where a key lives, the copies that survive a zone, and what one machine does with a write.

Where a key lives

Illustrative exampleNumbers here are illustrative: round figures to reason with, not measurements.

Hash keys onto a ring and give each machine many small arcs of it (virtual nodes), as the Dynamo paper does (Dynamo paper (opens in a new tab)); Cassandra defaults to 16 tokens per machine (cassandra.yaml (opens in a new tab)). A key belongs to the machine whose arc it falls in. A 31st machine takes a few arcs from many machines, about a 31st of the data; the other keys stay put. Never place keys by hash(key) mod the machine count when machines change: going from 30 to 31 moves about 30 of every 31 keys.

Fixed partitions mapped by a small table, rendezvous hashing, or ranges that split and move, as in Raft-based stores, work too. Every machine keeps the map, so any machine forwards a request to the key’s owners.

Ways to place keys

Compared on	Hash mod N	Ring, many arcs each	Fixed partitions	Ranges that split
A 31st machine moves	About 30 of every 31 keys	About a 31st, from many machines	Whole partitions, about its share	The ranges moved to it
Finding a key	Arithmetic	The ring in memory	A small table	A range map, cached

A client writes cart:42 to any machine, which hashes the key, finds its arc and forwards the write to the owners.
A 31st machine joins and takes a few small arcs from many machines.
A retired machine’s arcs pass to many different machines, a little each.

Architecture diagramWhere a key lives

Read:Learn: Partitioning, sharding and consistent hashing Learn: Apache Cassandra

Try it: Where a key livesGuided practice · Distributed key-value store, Shared key-value store · Senior

Keep reading+1 min

Ranges for scans, hashes for spread

Hashing spreads keys evenly but loses their order, so range scans would touch every machine. Stores that need scans keep sorted ranges and split them when they grow, at the cost of watching for a range that everyone writes to. This brief has no scans, so hashing is the simpler fit.

Check yourself

Why many small arcs per machine rather than one?
Show answerHide answer

Copies that survive a zone

Illustrative exampleNumbers here are illustrative: round figures to reason with, not measurements.

Keep three copies of each key. Walking clockwise from the key’s arc, take the next machines but skip positions on a machine already chosen, as the Dynamo paper’s preference list does (Dynamo paper (opens in a new tab)), and skip zones too: Cassandra’s NetworkTopologyStrategy places copies on different racks, which can be set to the zones (Cassandra docs (opens in a new tab)).

A put goes to all three copies, and the coordinator says “saved” once two, in different zones, have recorded it. Losing any machine or any zone then leaves a copy with every acknowledged write; the third copy catches up later. A Raft group per range with a member in each zone, acknowledging on a majority, gives the same guarantee.

A put for cart:42 reaches copies in zones A, B and C; A and B record it, and the client hears “saved”.
Zone B loses power just after: the copy in zone A still has the write, and new writes succeed on A and C.
A get from zone A a second later reaches copies that are up; what it sees depends on how many it asks.

Architecture diagramCopies that survive a zone

Inside one machine: from put to disk and back

Illustrative exampleNumbers here are illustrative: round figures to reason with, not measurements.

A put that reaches a copy is appended to the commit log first, then inserted into the memtable, a table in memory kept sorted by key, and only then acknowledged; after a crash, the machine replays the log (Cassandra docs (opens in a new tab)). The log reaches the disk before the reply in batch mode, or every 10 seconds in periodic mode, the default, where the copy in another zone covers those seconds.

When a memtable fills, it is written out as an immutable SSTable, sorted by key, and compaction merges SSTables in the background, keeping the newest version of each key. A get checks the memtable, then SSTables from newest to oldest; a bloom filter per SSTable says “not here” for most files without reading them (Cassandra docs (opens in a new tab)). A B-tree with a write-ahead log, or an embedded RocksDB, is also a sound choice when justified.

Ways to store keys on one machine

Compared on	Hash map and hourly snapshots	LSM tree	B-tree with a log
A restart after a reply	Loses up to an hour	Replays the log	Replays the log
Writes	Fast, in memory

Record	Key fields	Why it exists
Item	namespace, key, value, version, `expires_at`, deleted	The value, and what copies compare
Namespace	team, copies, default read and write, limits	Each team’s settings and limits

From	To	Cause
current	superseded	A newer version is written
current	deleted	A delete marker is written, or the expiry passes
superseded	removed	Dropped when the machine merges its files
deleted	removed	Dropped once every copy has seen the delete

Call	Returns	Notes
`GET /v1/{namespace}/keys/{key}?read=latest\|fast`	The value, its version and expiry	“latest” asks enough copies to overlap the ones that took the last save
`PUT /v1/{namespace}/keys/{key}`	The new version	`ttl`, `write=default\|fast`, and `if_absent` or `if_version`; `412` when the condition fails
`DELETE /v1/{namespace}/keys/{key}`	`204`	Writes a delete marker
`POST /v1/namespaces`	The namespace	Admin only: copies, defaults and limits

Design a distributed key-value store

The answer in two minutes

Also asked as

What interviewers look for

The decisions that matter

Understanding the problem

Functional requirements

Non-functional requirements

Scale

Out of scope

The numbers that shape the design

A year of growth

What a stored value carries

The store’s API

High-level design

Where a key lives

Ranges for scans, hashes for spread

Copies that survive a zone

Inside one machine: from put to disk and back

Sign in with Google to read the rest

Practise this

The same guarantee with consensus

Batch or periodic sync

The answer in two minutes

Also asked as

What interviewers look for

The decisions that matter

01Understanding the problem

Functional requirements

Non-functional requirements

Scale

Out of scope

02The numbers that shape the design

A year of growth

03What a stored value carries

04The store’s API

05High-level design

Where a key lives

Ranges for scans, hashes for spread

Copies that survive a zone

Inside one machine: from put to disk and back

Sign in with Google to read the rest

12Practise this

The same guarantee with consensus

Batch or periodic sync

Understanding the problem

The numbers that shape the design

What a stored value carries

The store’s API

High-level design

Practise this