Key-Value Store at Scale

A key-value store (like Redis, DynamoDB, or Memcached) maps keys to values. It's the simplest possible data model. No schemas, no joins, no SQL. Just PUT(key, value) and GET(key). This lesson introduces the foundational distributed systems concepts: partitioning, replication, and consistency, the building blocks behind every scalable database.

Step 1: Key-Value Store Requirements

Key-Value Store Functional Requirements

• PUT(key, value): store a value under a key (overwrite if exists)
• GET(key): retrieve the value for a key (return null if not found)
• DELETE(key): remove a key-value pair
• Keys are strings (max 256 bytes), values are blobs (max 10 KB)
• Optional TTL (time-to-live) per key

Key-Value Store Non-Functional Requirements

Key-Value Store Scope Exclusions

• Range queries (that's a sorted store like Cassandra)
• Transactions across multiple keys
• Secondary indexes
• Complex data types (lists, sets, sorted sets, that's Redis)

Step 2: Key-Value Store Estimation

Key-Value Store Traffic

• 500K reads/sec + 100K writes/sec = 600K ops/sec total
• A single server handles ~50K ops/sec → need at least 12 servers for throughput

Key-Value Store Storage

• 100 TB of data
• At 64 GB usable RAM per server → about 1,600 servers for in-memory store
• At 2 TB SSD per server → about 50 servers for disk-based store

Which number matters? Storage is the binding constraint. We need at least 50 servers to hold 100 TB on disk. Those 50 servers also provide 50 × 50K = 2.5M ops/sec, far more than the 600K we need. So 50 servers is the starting point, with replication (3×) bringing it to about 150 nodes total.

Key-Value Store Key Insight

The challenge is distributing 100 TB across many machines while keeping reads fast, writes durable, and the system available during failures. The data model is trivial. The distributed systems problems are not.

Step 3: Key-Value Store High-Level Design

┌─────────────────────────────────────────────────────────────────┐
│                      KEY-VALUE STORE                            │
├─────────────────────────────────────────────────────────────────┤
│                                                                 │
│  ┌──────────┐     ┌──────────────┐                              │
│  │  Client  │────►│  Coordinator │                              │
│  │          │◄────│  (any node)  │                              │
│  └──────────┘     └──────┬───────┘                              │
│                          │  hash(key) → find node on ring       │
│              ┌───────────┼───────────┐                          │
│              ▼           ▼           ▼                          │
│       ┌──────────┐ ┌──────────┐ ┌──────────┐                    │
│       │ Node A   │ │ Node B   │ │ Node C   │  ...               │
│       │ (primary)│ │ (replica)│ │ (replica)│                    │
│       │ WAL +    │ │ WAL +    │ │ WAL +    │                    │
│       │ LSM Tree │ │ LSM Tree │ │ LSM Tree │                    │
│       └──────────┘ └──────────┘ └──────────┘                    │
│                                                                 │
│       Nodes arranged on a hash ring (consistent hashing)        │
│       Each key hashed → assigned to next node clockwise         │
│       Replicated to N-1 additional nodes for durability         │
└─────────────────────────────────────────────────────────────────┘

Three core problems to solve:

Note on the coordinator: In Dynamo-style systems (Cassandra, DynamoDB), any node can be the coordinator. The client connects to any node and it routes the request. In other systems (MongoDB), a dedicated router (mongos) handles coordination. Both work; the leaderless approach is simpler for key-value stores.

1. Partitioning: which node stores which keys?
2. Replication: how to keep copies for durability and availability?
3. Consistency: what happens when replicas disagree?

Step 4: Key-Value Store Deep Dive

Partitioning: Consistent Hashing

With 50 nodes, how do you decide where a key lives?

Naive approach: hash(key) % N

• Works until you add or remove a node
• Changing N remaps almost every key → massive data migration

Better: Consistent hashing

• Arrange nodes on a virtual ring (0 to 2^32)
• Hash the key → find the next node clockwise on the ring
• Adding/removing a node only affects its neighbors (~1/N of keys move)

        Node A
       /      \
    Node D     Node B
       \      /
        Node C

Key "user:123" → hash → lands between A and B → stored on B
Add Node E between A and B → only keys between A and E move to E

Virtual nodes: Each physical node gets 100-200 positions on the ring. This spreads keys more evenly. Without virtual nodes, some nodes get disproportionate load by chance.

Replication

Storing data on one node means losing it if that node dies. Replicate to N nodes (typically N=3):

Key "user:123" → primary: Node B, replicas: Node C, Node D
                 (next N-1 nodes clockwise on the ring)

Write path:

1. Client sends PUT to coordinator
2. Coordinator forwards to primary + replicas
3. Wait for W acknowledgments before responding (W = write quorum)

Read path:

1. Client sends GET to coordinator
2. Coordinator reads from R nodes
3. Compare responses, return the value with the highest timestamp (or vector clock)

What if replicas disagree? Each value is stored with a timestamp (or version vector). When R nodes return different values, the coordinator picks the one with the latest timestamp. The stale replica is repaired in the background (read repair).

Consistency: Quorum (W + R > N)

With N=3 replicas, you choose how many must acknowledge:

The rule: W + R > N guarantees you'll read the latest write. Because at least one node in the read set must have the latest write.

Example: N=3, W=2, R=2. A write goes to nodes A, B, C. Two acknowledge (say A, B). A read queries two nodes (say B, C). Node B has the latest value → client gets it.

Handling Failures

Node goes down temporarily:

• Hinted handoff: writes meant for the dead node go to a neighbor with a "hint." When the node recovers, hints are replayed.

Node goes down permanently:

• Anti-entropy: background process compares replicas using Merkle trees (hash trees). Detects and repairs inconsistencies.

Network partition:

• AP system (like DynamoDB): both sides continue serving reads/writes. Resolve conflicts later (last-write-wins or vector clocks).
• CP system (like ZooKeeper): minority side stops serving. Majority side continues. Guarantees consistency but sacrifices availability. Note: ZooKeeper is a coordination service (leader election, locks, config), not a general-purpose KV store. It's CP because coordination requires strong consistency.

Storage Engine (Per Node)

Each node needs to store key-value pairs on disk efficiently:

For our KV store: LSM Tree, optimized for high write throughput (100K writes/sec). Writes go to an in-memory buffer, flushed to disk in sorted batches. Reads check memory first, then disk files.

Compaction: Over time, LSM Trees accumulate many SSTables on disk. Reads get slower because each GET must check multiple files. Compaction merges SSTables in the background, combining files, removing deleted keys (tombstones), and discarding overwritten values. This keeps read performance stable.

Write-Ahead Log (WAL)

What if the node crashes before flushing memory to disk?

Write arrives → append to WAL (disk) → write to memtable (memory) → ack client

Node crashes → restart → replay WAL → memtable restored

The WAL is an append-only file. Sequential writes are fast. It guarantees durability even if the in-memory data is lost.

Deletes in an LSM Tree (Tombstones)

You can't just remove a key from an SSTable. SSTables are immutable files. Instead:

1. Write a tombstone marker: DELETE("user:123") → store a special "deleted" record
2. Reads see the tombstone and return "not found"
3. During compaction, the tombstone and all older values for that key are physically removed

This means deleted keys still occupy space until compaction runs. It also means a GET must check if the latest entry is a tombstone, which is slightly more complex than just "key not found."

Data Flow Summary

PUT("user:123", "{name: alice}"):
  1. Client → Coordinator
  2. Coordinator → hash("user:123") → determines Node B is primary
  3. Forward to Node B + replicas (C, D)
  4. Each node: append WAL → write memtable → ack
  5. Coordinator waits for W=2 acks → responds to client

GET("user:123"):
  1. Client → Coordinator
  2. Coordinator → hash → Node B is primary
  3. Read from R=2 nodes (B, C)
  4. Compare responses → return latest value

Key Takeaways

• Consistent hashing distributes keys with minimal disruption

  Adding or removing a node only moves ~1/N of keys (not all of them like hash % N). Virtual nodes ensure even distribution across physical machines.

• Replication provides durability and availability

  N=3 replicas means any single node can die without data loss. The tradeoff is write amplification (every write goes to 3 nodes) and potential consistency issues between replicas.

• W + R > N guarantees strong consistency

  The quorum formula ensures at least one node in every read set has the latest write. Tuning W and R lets you trade consistency for speed depending on your use case.

• LSM Trees optimize for write-heavy workloads

  Sequential writes to a WAL + in-memory buffer + periodic flush to sorted files. This is why Cassandra and DynamoDB handle 100K+ writes/sec. They never do random disk writes.

• CAP theorem forces a choice during partitions

  When the network splits, you either stay consistent (reject writes on the minority side) or stay available (accept writes everywhere, resolve conflicts later). Most KV stores choose AP for availability.

• The data model is simple, the distributed systems are hard

  PUT/GET/DELETE is trivial. Partitioning, replication, consistency, failure handling, and compaction are where the complexity lives. This is why "design a key-value store" is a popular interview question.