Design problems
Classic building blocksHardSDE-3 40 min

Design a distributed key-value store (like DynamoDB)

Asked at
AmazonFlipkartGoogleLinkedInMicrosoftPhonePe

The interviewer asks

“Design a distributed, durable key-value store that supports put(key, value) and get(key), stores hundreds of terabytes, stays available when machines and even a whole availability zone fail, and lets clients choose between fast and strongly consistent reads.”

What a strong answer looks like

A strong answer builds the store from well-known parts and justifies each. Partition keys with consistent hashing (virtual nodes), replicate each key to N=3 nodes across zones, and choose quorum settings (W and R) to trade consistency for latency. Each node stores data with a log-structured merge tree (write-ahead log, memtable, SSTables, compaction, Bloom filters), which suits write-heavy workloads. Then explain how the system stays correct under failure: hinted handoff for short outages, Merkle-tree anti-entropy for long ones, gossip for membership, and conflict handling with versions (last-write-wins vs vector clocks). Senior candidates contrast this leaderless Dynamo design with a leader-per-partition design (DynamoDB today, using Multi-Paxos) and explain when each fits.

Concepts to mention

consistent hashing with virtual nodes / partition splitsreplication factor N=3 across AZsquorum R + W > Nsloppy quorum and hinted handoffleader per partition via Raft/Paxos vs leaderlessLSM tree: WAL, memtable, SSTables, compaction, Bloom filterstombstones and deletesMerkle-tree anti-entropy and read repairconflict resolution: LWW vs vector clocksgossip-based failure detection

1 Requirements

Functional

  • put(key, value) and get(key) with values up to 400 KB
  • delete(key)
  • Per-request consistency choice: eventually consistent read or strongly consistent read
  • Conditional writes (put only if version = v) for optimistic concurrency
  • Optional TTL per item

Non-functional

  • Latency p99: under 10 ms for reads and writes within a region
  • Availability 99.99%; survive loss of any node and of one availability zone
  • Durability: no acknowledged write is lost (data on at least 2 zones before ack)
  • Scale: 500 TB of data, 1M reads/s and 300k writes/s, growing 2× per year
  • Elastic: add nodes without downtime; load automatically rebalances

Out of scope (say it out loud)

  • Secondary indexes and range queries over non-key fields (mention global secondary indexes as an extension)
  • Multi-item transactions across partitions
  • SQL query layer

2 Estimates

Raw storage with replication~1.5 PB500 TB × 3 replicas = 1.5 PB; plus ~30% for compaction space and growth ≈ 2 PB of disk
Nodes~250 storage nodes8 TB usable NVMe per node: 2 PB ÷ 8 TB = 250 nodes, spread across 3 AZs
Per-node load~12k reads/s + 3.6k writes/sReads: 1M × (R=1..2 replicas) ≈ 1.5M replica reads ÷ 250 ≈ 6k-12k/s; writes: 300k × 3 replicas ÷ 250 ≈ 3.6k/s
Partitions~50,000 partitions of ~10 GB500 TB ÷ 10 GB; small partitions make moves fast (10 GB at 200 MB/s ≈ 50 s)
Write bandwidth per node~40 MB/s incl. compaction3.6k writes/s × ~2 KB ≈ 7 MB/s, × ~5 for LSM write amplification ≈ 36 MB/s; fine for NVMe

3 API

EndpointWhat it does
PUT/tables/{t}/items/{key}Body: value, optional condition (expected_version), optional ttl. Returns new version.
GET/tables/{t}/items/{key}?consistent=true|falseEventually consistent (any replica, cheaper) or strongly consistent (leader or quorum) read.
DELETE/tables/{t}/items/{key}Writes a tombstone; data is removed later by compaction.
POST/tablesCreate table with partition key schema and capacity mode.
gRPCInternal.Replicate(key, value, version)Node-to-node replication and hinted handoff.

4 Data model

ItemLSM tree on each storage node (RocksDB-like)
partition keyvalue bytesversion (or vector clock)ttltombstone flag

Append-only writes to WAL + memtable; flushed to sorted immutable SSTables; compaction merges and drops overwritten values and expired tombstones.

Partition mapMetadata service (Paxos/Raft group, e.g. etcd)
token range → partition_id → replica nodes (leader, followers)epoch

Small, strongly consistent, cached by request routers; changes on splits and moves.

MembershipGossip protocol
node_idaddressstatusheartbeat counter

Each node gossips with a few random peers per second, so failure detection scales without a central bottleneck.

Merkle treesIn memory per partition replica
hash tree over key ranges

Replicas compare root hashes and walk down only where they differ, to sync missing data cheaply.

5 Architecture

CLIENTSEDGESERVICESASYNCDATAget/putcached partition mapto leader/coordinatorreplicatereplicatereplica downreplay latercompare treessplit/movesnapshots + WALClient SDKRequest routersPartition metadata(Raft)Rebalancer /controllerAnti-entropy (Merkle)Hinted handoff queueStorage node AZ-b(LSM)Storage node AZ-a(LSM)Storage node AZ-c(LSM)Backups (S3)
ClientEdge / gatewayService / workerQueue / streamCacheDatabase / storageAI / modelThird partyrequestasync (event / queue)

Hover a box to highlight its connections.

6 Request flows

Write with quorum (N=3, W=2)

  1. 1Client SDK hashes the key (e.g. MD5/murmur) to a token and asks a router.
  2. 2Router finds the partition and its 3 replicas (one per AZ) from the cached partition map.
  3. 3The coordinator (partition leader, or the first healthy replica in a leaderless design) assigns a version and appends the write to its WAL and memtable.
  4. 4It sends the write to the other two replicas in parallel.
  5. 5When W=2 replicas (including itself) have the write in their WAL, it returns success to the client (~3-5 ms across AZs).
  6. 6The third replica catches up asynchronously; if it is down, the write is kept as a hint and replayed when it returns.

Read

  1. 1Eventually consistent read: router sends the read to one replica (often the closest), which checks memtable, then SSTables using Bloom filters to skip files that cannot contain the key.
  2. 2Strongly consistent read: router sends to the partition leader (which has all committed writes) or reads R=2 replicas and returns the newest version.
  3. 3If replicas disagree in a quorum read, the coordinator returns the newest and writes it back to the stale replica (read repair).
  4. 4The SDK retries on another replica after a short timeout (hedged request) to cut tail latency.

7 Deep dives

Partitioning and rebalancing

Keys are hashed onto a ring. In the original Dynamo paper each physical node owns many virtual nodes (tokens), so load spreads evenly and a new node takes small slices from many peers rather than half of one neighbour. Modern systems such as DynamoDB and Cassandra-like stores often use explicit partitions instead: the key space is split into ranges of hashed keys, each partition is limited to around 10 GB (DynamoDB documents that a partition serves up to about 3,000 read units and 1,000 write units per second), and a partition is split when it grows too big or too hot. A controller then moves partitions to balance disk and traffic across nodes. Moving a 10 GB partition means: create a new replica, stream an SSTable snapshot, then tail the log until caught up, then switch the partition map version. Because routers cache the map, nodes must reject requests for partitions they no longer own with a 'wrong owner, epoch X' error, so routers refresh. Hashing keys spreads load but kills range scans across keys; DynamoDB offers a sort key inside a partition key to keep ordered data together, which is the right compromise.

Replication and quorums

With N replicas, writes wait for W acks and reads query R replicas. If R + W > N (e.g. N=3, W=2, R=2), every read set overlaps every write set, so a read sees the latest acknowledged write, assuming no failures move writes to other nodes. This is why N=3, W=2, R=2 is the classic choice; W=1, R=1 is fastest but only eventually consistent. Leaderless quorums (Dynamo, Cassandra) stay writable during failures using sloppy quorums: if a home replica is down, another node accepts the write with a hint. That boosts availability but breaks the R + W > N guarantee during the failure, a point many candidates miss. The alternative is a leader per partition chosen by Multi-Paxos or Raft (DynamoDB, Spanner, CockroachDB): writes go to the leader, are committed when a majority (2 of 3) has them in the log, and strongly consistent reads go to the leader. This gives linearizable per-key operations and simple conditional writes, at the cost of leader elections (a few seconds of unavailability for one partition) and leader hotspots. For a product that offers conditional writes and 'consistent read' options, the leader-based design is easier to make correct.

Storage engine: LSM trees

Each node needs fast writes and decent reads on NVMe. An LSM tree appends each write to a write-ahead log (sequential, fsync batched every few ms), inserts it into an in-memory sorted memtable, and when the memtable reaches ~64-128 MB flushes it to an immutable sorted SSTable. Reads check the memtable, then SSTables from newest to oldest; per-file Bloom filters (around 10 bits per key gives ~1% false positives) avoid touching files that cannot contain the key, and a block cache keeps hot blocks in memory. Background compaction merges SSTables, discarding overwritten versions and tombstones older than a grace period. Leveled compaction keeps read amplification low (good for read-heavy tables) at the cost of more write amplification; size-tiered compaction is cheaper for writes but uses more disk and more files per read. Deletes are tombstones: they must live long enough to reach all replicas, otherwise a replica that missed the delete can bring the data back during repair. B-trees (InnoDB, Postgres) are the alternative: better for read-heavy, in-place updates, worse for heavy random writes.

Conflicts and versioning

With leaderless replication, two clients can write the same key on different replicas during a partition. Options: last-write-wins using timestamps (simple; Cassandra does this; but clock skew can silently drop a newer write), or vector clocks that detect concurrent writes and return both siblings to the client to merge (the original Dynamo shopping cart). Vector clocks are correct but push complexity to every client, which is why most products avoid them. With a leader per partition, the leader assigns a monotonically increasing version per key, so there are no siblings, and conditional puts (expected_version) give safe read-modify-write. Recommend the leader design for a general-purpose product, and mention CRDTs (counters, sets) for the cases where merge semantics are natural.

8 Trade-offs

Leaderless vs leader-based replication

Leaderless (Dynamo, Cassandra) gives always-writable behaviour and no election pauses, but conflict handling and weaker guarantees. Leader-based (DynamoDB today, Spanner, TiKV) gives strong per-key consistency and simple conditional writes, but a partition is briefly unavailable for writes during leader failover. Choose based on whether the product needs 'consistent read' and compare-and-set, which most do.

LSM tree vs B-tree

LSM trees turn random writes into sequential ones, compress well and handle heavy writes; they pay with read and write amplification and compaction tuning. B-trees give predictable reads and in-place updates but suffer under heavy random writes. A KV store for mixed, write-heavy workloads usually chooses LSM (RocksDB).

Hash partitioning vs range partitioning

Hashing spreads load evenly and avoids hot ranges from sequential keys (timestamps), but loses ordered scans. Range partitioning (Bigtable, HBase, Spanner) supports scans but needs automatic splitting and can create hot spots for monotonic keys. A composite key (hash partition key + sorted sort key) gives most of both.

9 Bottlenecks

Hot partition

One key or one partition getting 50k ops/s exceeds a single replica set. Split hot partitions by key range, cache hot reads in front (DynamoDB Accelerator-style), and advise write sharding (append a random suffix to the key and aggregate on read) for counters.

Compaction falling behind

Under heavy writes, compaction debt grows, reads touch more files and disk fills. Monitor pending compaction bytes, throttle writes (backpressure) when debt is too high, and add nodes before disks pass ~70%.

10 Failure handling

Single node failure

Gossip or the controller detects it in seconds. In a leader design, partitions it led elect new leaders among the 2 remaining replicas (still a majority). Hints accumulate for short outages; if the node is gone for good, the controller creates new replicas elsewhere from the healthy ones.

Whole availability zone fails

Each partition has one replica per AZ, so 2 of 3 replicas remain, which is still a majority for W=2 and leader election. Capacity drops by a third, so normal load must stay under ~65% of total to survive. When the AZ returns, replicas catch up via logs or Merkle-tree repair.

Silent data divergence

Bit rot or missed writes leave replicas different. Checksums on every block catch corruption on read, and periodic Merkle-tree anti-entropy finds and repairs differing key ranges without comparing every key.

11 What impresses

  • Explains that sloppy quorums break the R + W > N guarantee during failures
  • Contrasts Dynamo-paper design with how DynamoDB actually works today (leader-based, Multi-Paxos)
  • Describes how a partition moves safely using map epochs and wrong-owner errors
  • Knows Bloom filter sizing and LSM compaction trade-offs
  • Mentions tombstone grace periods and the 'zombie data' problem
  • Capacity planning for AZ loss (run under ~65% so 2 AZs can carry the load)

12 Mistakes

  • Saying 'use consistent hashing' and stopping there, without replication or failure handling
  • Claiming R + W > N gives strong consistency in all cases
  • Ignoring how data moves when a node is added
  • Using last-write-wins without mentioning clock skew
  • Not describing the per-node storage engine at all
  • Putting all replicas in the same AZ

13 Follow-ups

Q1How would you add global secondary indexes?
Q2How do you implement conditional writes correctly?
Q3How would you support multi-region replication like DynamoDB Global Tables?
Q4What happens to in-flight writes during a partition move?
Q5How do you back up and restore to a point in time?
Q6How do you handle a key that receives 100k writes per second?

Saved on this device. Create a free account to keep it on every device.

Practise this with Sam

A real mock interview, out loud, with an honest scorecard. Free to start.

Start free