Skip to main content

Command Palette

Search for a command to run...

Design a Key-Value Store | System Design Question

Updated
17 min readView as Markdown

"Design a key-value store" is one of the most common system design interview questions, and it is a great one because building it touches almost every idea in distributed systems. We will take it end to end on the 4-step framework: scope it, sketch a high-level design, go deep, then wrap up. Everything here is in plain language, and every term is defined the first time it appears - if a word shows up undefined, that is a mistake in the writing.

A key-value store keeps data as simple key-to-value pairs. You put(key, value) to store something and get(key) to read it back - like a coat check, where a ticket (the key) gets you your coat (the value). The value is an opaque blob: the store does not look inside it, it just hands it back exactly as you gave it. Redis, Amazon DynamoDB, and Apache Cassandra are all key-value stores. On a single machine this is trivial; the whole interview is about making it work across many machines without ever losing data or going down. (One word up front: a node, or server, just means one computer in our system - we will have many.)

Step 1 - Scope the problem

Never start designing before you know the requirements. In an interview you ask; here is roughly how it goes. (If your mind blanks when the question lands, this post is about exactly that.)

You: How big is one value - a few bytes, or huge files? Interviewer: Small. Under 10 KB each.

You: How much data in total? Will it fit on one machine? Interviewer: You tell me. Assume we are a big product. You: Then I will assume it is far too much for one machine, so this has to be spread across many machines. I will design for that.

You: When part of the system can't be reached, would you rather it keep answering (even if the answer is slightly out of date) or stop answering until it is sure it is correct? Interviewer: Keep answering. Staying up matters.

You: Should it also do search, or transactions across many keys? Interviewer: No. Just get and put on a single key.

Requirements

Functional (what it must do):

  • get(key) and put(key, value), one key at a time.
  • Store far more data than one machine can hold.

Non-functional (how well it must do it):

  • Available - keeps answering even when some machines are down.
  • Scalable - we can add machines as we grow, with no downtime.
  • Fast - a get or put returns quickly.
  • Durable - once stored, data is never lost, even if a machine dies.

A quick estimate

Say 1 billion keys, up to 10 KB each: that is up to ~10 TB of data, and with 3 copies of everything (we will see why in a moment) call it ~30 TB. No single machine holds that, which confirms this must be distributed. Each get/put is cheap on its own, so the hard part is not raw compute - it is spreading the data across machines and keeping the copies in agreement. That is where the whole design goes.

Step 2 - High-level design

I will build this up piece by piece rather than draw the final picture cold.

Start with one machine, then break it. On one computer, a key-value store is just a big lookup table held in memory (memory meaning RAM, the fast storage a running program uses), where each key points to its value. put adds an entry, get reads one - both instant. But one machine hits two walls: it runs out of space (a big product has more data than any one machine's disk, "disk" being the slower storage that survives a restart), and it is a single point of failure (if it dies, everything is gone and the service is down). So I have to spread the data across many machines. That immediately raises two questions - which machine holds which key, and how do I not lose a key if its machine dies - plus one trade-off I cannot dodge.

The trade-off (CAP). My machines talk over a network, and networks break, so sometimes some machines cannot reach others - call it a network split. During a split, a request forces a choice I cannot avoid: return the most recent value (consistency) or return an answer at all (availability). If a value lives on both sides of the break, a write reaches one side but not the other, so a reader on the far side either gets the old value (I kept availability) or gets refused until the break heals (I kept consistency). I cannot have both while the wire is cut - that is the CAP trade-off. Per Step 1, I choose available: for a cart, a profile, or a feed, briefly stale data beats being down. That single choice, the same one Dynamo and Cassandra made, drives everything below.

Which machine holds a key? The obvious rule - number the machines and send a key to hash(key) mod N (a hash function is math that turns any input into an evenly-spread number; "mod N" is the remainder, which picks a machine) - falls apart the moment I add or remove a machine, because the remainder changes for almost every key at once and nearly all the data has to move. The fix is consistent hashing: place the machines and keys around a circle and store each key on the first machine clockwise from it.

A ring of servers s0 through s7 with a key placed on it; walking clockwise from the key's position, the first server reached is the one that stores it

Now adding or removing a machine only moves the keys in one small stretch of the circle; everything else stays put. That is what makes growing the cluster cheap. There is more to it (like keeping the load even), and it has its own full walkthrough: What is consistent hashing?. For here, the point is: it maps each key to a machine, and resizing only shuffles a little data. The first machine clockwise from a key is its home node.

Don't lose data: replicate. One copy on one machine means one crash loses it. So I pick a number N (say 3) and store each key on N machines: its home node plus the next N-1 clockwise. Those copies are replicas (a replica is a full copy on another machine).

A key stored on three replicas: its home node plus the next two machines clockwise, giving three copies of the same data

For real safety I put those copies in different data centers (a data center is a building full of servers), so one power cut cannot kill all three.

Keep the copies in agreement: quorums. First a word I will use from here on: the coordinator is simply whichever node is handling your current request - it takes your get/put, talks to the replicas for you, waits for enough of them, and replies. It is not a separate machine; any node can play the role, and often it is the key's own home node. Now, with 3 copies, how many must respond for an operation to count? That is a quorum, three numbers: N (copies per key, here 3, call them A, B, C), W (how many must confirm a write), and R (how many must answer a read). The coordinator contacts all replicas; W and R just say how many replies it waits for. A write with W=2 returns as soon as 2 of the 3 confirm; a read with R=2 waits for 2 answers and returns the newest.

The useful rule is W + R > N. Here is why it works, and it is just counting:

With N=3, a write with W=2 reaches A and B while C lagged, and a read with R=2 asks B and C; node B is in both groups, so it holds the new value and the read is guaranteed to see it

The write touched at least 2 machines, the read hears from at least 2, and there are only 3 - so the two groups must share one (here B). B has the new value and B is in the read, so the read cannot miss the latest write. That is what 2 + 2 > 3 guarantees. With W=1, R=1 the write and read can land on different machines (1 + 1 is not > 3), so reads can be stale. Wait for more replies and you are slower but safer; wait for fewer and you are faster but riskier - W=2, R=2 is the balanced middle.

The architecture so far. Every machine runs the same jobs, so there is no special master and no single point of failure; any machine can coordinate a request to the replicas on the ring:

The full picture: a client sends get and put to a coordinator, which is any node on a ring of equal machines built with consistent hashing; each key is copied to N machines, and reads and writes use the R and W quorums

That is a working store. The deep dive is where it gets interesting: what happens when two people write at once, and what happens when machines fail.

Step 3 - Deep dive

Concurrent writes: conflicts and vector clocks

Because I chose "keep answering," two people can update the same key at the same moment on different machines, and the copies genuinely disagree. Two quick terms: strong consistency means a read always returns the very latest write (a bank balance needs this); eventual consistency means a read might briefly be a little stale, but if writes stop, all copies soon agree (a "likes" count off by one for a second is fine). Strong consistency makes machines wait on each other, which loses the availability I promised, so like Dynamo and Cassandra I accept eventual consistency - which means handling those disagreements.

You might think consistent hashing prevents this, since each key has a fixed home node. It does not: knowing which machines hold a key does not order its writes. Those N machines are equal peers, none of them a boss that decides the official order, so two near-simultaneous writes (or writes accepted on different machines during a network split) leave the copies holding different values. That difference is the conflict.

To sort it out, each value carries a small version stamp called a vector clock - just a tally attached to one key, counting how many times each machine has written that key. Sx=2, Sy=1 means "machine Sx wrote this key twice, Sy once." Every key has its own tally, and when a machine handles a write (the first put or a later update), it adds one to its own number. Comparing two stamps: if one's numbers are all less-than-or-equal to the other's, the second is simply newer (keep it); if each has a number the other lacks, each side saw a write the other missed - a real conflict.

A shopping cart tracked with version stamps: milk then eggs on machine Sx (Sx=1, Sx=2), then at the same time bread on Sy and butter on Sz, producing two conflicting versions because neither stamp covers the other, and finally the client merges both into one cart with all four items

Milk then eggs both go through Sx (its count climbs 1, 2 - a clean line). Then at once, bread on Sy (Sx=2, Sy=1) and butter on Sz (Sx=2, Sz=1): the bread stamp has a Sy the butter one lacks, and vice versa, so neither covers the other - a conflict. The system does not silently drop a grocery item; on the next read it returns both carts, the app merges them (combine into one cart with all four items), and writes the merged result. The comparison in code is just a per-machine check:

type VectorClock = Map<string, number>; // machine id -> its write count for this key

// True if clock A is "covered by" B - i.e. B is newer, with no conflict.
function isCoveredBy(a: VectorClock, b: VectorClock): boolean {
  for (const [machine, countA] of a) {
    const countB = b.get(machine) ?? 0;
    if (countA > countB) return false; // A is ahead somewhere, so B does not cover it
  }
  return true;
}

// If neither covers the other, the two versions conflict (must be merged).
function inConflict(a: VectorClock, b: VectorClock): boolean {
  return !isCoveredBy(a, b) && !isCoveredBy(b, a);
}

The simpler alternative, last-write-wins (keep whichever has the later timestamp), is less work but would quietly drop either the bread or the butter. Vector clocks are the price of never losing a write.

Knowing which machines are down: gossip

Before I can handle a failure I have to detect it. The coordinator finds out a machine is dead when it times out reaching it - but that only helps that coordinator, that once. Since any node can coordinate the next request, if only the machine that just timed out knew, every other node would rediscover the failure the slow way. So the machines tell each other who is up (there is no central monitor, and one machine's word is not enough - its own link may be the broken one).

They share it with a gossip protocol, modeled on how a rumor spreads. Each machine keeps a list with a heartbeat number per machine (a counter that says "I am alive"). Periodically each one bumps its own heartbeat and sends its list to a few random machines, which merge in anything newer and pass it on. In a few rounds a fresh heartbeat reaches everyone; a heartbeat that stops advancing means that machine is down.

Gossip: a machine bumps its own heartbeat and sends its list to a few random machines, which merge it and pass it on; a heartbeat that stops advancing is treated as down, and that news spreads the same way

Gossip beats having every machine check every other one directly, which floods the network as the cluster grows; each machine talks to only a few, yet the news still reaches all quickly.

A machine that comes back: hinted handoff

Say I am writing a key whose replicas are A, B, C, and gossip says C is down. I do not want the write to fail - staying available is the point. So the coordinator writes A and B normally, and for C's copy it picks another healthy machine (call it D) and stores the data there with a hint attached - a note that says "this really belongs to C; D is only holding it for now." The write has its 3 copies and succeeds. When gossip says C is back, D hands the copy off to C and deletes its own. A copy held with a hint and later handed off is a hinted handoff.

The coordinator writes a key to A and B, but C is down, so it stores C's copy on another machine D with a hint that it belongs to C; when C returns, D hands the copy off to C and deletes its temporary copy

A machine that is gone for good: anti-entropy and Merkle trees

Hints cover a machine down for minutes. But if C's disk dies and a blank machine replaces it, or a copy quietly drifted, two machines that should be identical now differ, with no hints to replay. I have to find the differing keys and copy only those. That background repair is called anti-entropy ("entropy" being the gradual drift that creeps in; anti-entropy cleans it up).

Comparing whole datasets would ship terabytes to find a few differences. The tool that avoids this is a Merkle tree - a tree of fingerprints, where a fingerprint is a hash (a short number from a hash function; identical data gives an identical number, one changed byte changes it). Split the keys into buckets, hash each bucket (the bottom of the tree), hash each pair upward, until one root hash remains - a fingerprint of the whole dataset. Two machines compare by swapping hashes from the root down:

Comparing two machines with a Merkle tree: the root fingerprints differ, so the matching left half is skipped and only the right half is explored, down to the single bucket that differs, and only that bucket's keys are exchanged and repaired

If the roots match, the datasets are identical after swapping one number. If they differ, compare the two halves and descend only the side that disagrees, skipping matching branches, straight down to the exact bucket that differs - and copy only those keys. The data moved depends on how much the machines differ, not how much they hold. This is what Cassandra's repair does. (A whole data center going dark is already covered by keeping copies in different buildings.)

Optional: how one machine stores data on its disk

One level deeper, and only whiteboard it if the interviewer asks - this is how a single machine saves bytes, not part of the distributed design. A write is appended to a commit log (an append-only file, so nothing is lost on a crash), updated in a small in-memory table called the memtable, and when that fills up it is written to disk as an SSTable - a Sorted String Table, just a file of key-value pairs kept sorted by key and never changed once written. Because updates keep making new SSTables, a key could be in any of several files; each file gets a bloom filter (a tiny structure that says "definitely not here" or "maybe here") so a read skips the files that cannot hold the key - see what is a bloom filter.

Follow-up questions to expect

  • Why not strong consistency? It makes machines wait on each other, losing the availability and speed we were asked for. For a specific operation that needs the latest data, set W + R > N.
  • What if two writes truly conflict? Vector clocks catch it and we merge (or fall back to last-write-wins and accept losing one). We do not pretend conflicts cannot happen.
  • A single hot key overloads its machines? Consistent hashing spreads many keys evenly, but one red-hot key still lands on the same few machines; cache it or split it, like any hot key.
  • How does a machine join or leave with no downtime? Consistent hashing moves only one stretch of the ring, and gossip spreads the change, so the cluster rebalances quietly while serving requests.
  • Where is the single point of failure? There isn't one - every machine does the same jobs and any machine can coordinate.

Wrap up

The design in one breath: an available, distributed key-value store. Consistent hashing maps keys to machines and makes growth cheap; replication keeps N copies so a dead machine loses nothing; quorums (N, W, R) read the latest value while trading speed against freshness; vector clocks detect and merge conflicting writes; gossip, hinted handoff, and Merkle-tree repair keep it running and in sync through failures; cross-data-center copies survive a building going dark. That is, almost exactly, Amazon Dynamo and Apache Cassandra. The single idea underneath: we chose to stay available when the network breaks, and every mechanism here is either a benefit of that choice or the cleanup it requires. For a different design walked the same way, see design a URL shortener.

More from this blog

P

Programming, System Design and AI | Latencot

53 posts

Here, you will read all about tech and tech updates. From writing code to building AI systems, we'll talk and share about everything.