Distributed Key-Value Store (DynamoDB / Cassandra)

0) Problem Restatement

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.

1) Requirements

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

• PUT /kv/{key} with the value, and header consistency: one|quorum|all
• GET /kv/{key} → value + version
• DELETE /kv/{key}

2) High-Level Architecture

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

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")]

3) Storage Engine on Each Node

Most large KV stores use an LSM tree (Log-Structured Merge tree), which is great for heavy writes:

1. Commit log: append the write to a file on disk first (so it survives a crash).
2. Memtable: put it in a sorted in-memory table.
3. When the memtable is full, write it to disk as an immutable sorted file (SSTable).
4. Background compaction merges SSTables, dropping overwritten and deleted values.
5. 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.

4) 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).

• 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.

5) 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.

6) Deep Dive — Getting to 1M QPS and hot keys

1. Start: one node, about 80K ops/sec. Add caching for reads and tune the storage engine.
2. Replicate for reads: followers serve reads, but writes are still limited to one node.
3. Partition: consistent hashing across ~20 partitions spreads writes, and each partition has 3 replicas.
4. Smart clients: route directly to the owner, which saves one network hop.
5. Hot keys: if one key gets 100K reads/sec, cache it in the client or a front cache, or spread it across replicas with R=1 reads. For hot writes, split the key into sub-keys and combine them on read.
6. Rebalancing: add nodes gradually. Virtual nodes move small slices while serving traffic, and throttle the data transfer so users don't notice.

7) Trade-offs & Alternatives

8) Common Follow-up Questions

• "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.

9) Wrap-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.