Design a Key-Value Store

A key-value store maps a key to a value and supports put(key, value) and get(key). A single-node store is a hash table with persistence. A distributed key-value store spreads data over many machines to exceed one machine's capacity and to survive failures. This article designs a Dynamo-style store: partitioned, replicated, always writable, and eventually consistent by default, with tunable consistency per request. Systems in this family include Amazon Dynamo (the published design), Apache Cassandra, and Riak.

Requirements

Functional requirements

• put(key, value), get(key), delete(key).
• Keys are opaque byte strings, up to a few hundred bytes. Values are blobs up to about 1 MB.
• Optional per-request consistency level.

Non-functional requirements

• High availability. Writes and reads succeed during node failures and network partitions, where possible.
• Horizontal scalability. Capacity and throughput grow by adding nodes, with automatic redistribution of data.
• Low latency. Single-digit milliseconds at the median for a request within a datacenter.
• Durability. Acknowledged writes survive node crashes.
• Tunable consistency. Choose between low latency and stronger consistency per operation.
• Decentralized operation. No single coordinator node whose failure stops the system.

In terms of the CAP theorem, this design favors availability during partitions (AP) and accepts that replicas may briefly disagree.

Estimation

Assumptions: 100 TB of unique data, an average value of 10 KB, a replication factor of 3, 1 million requests per second at peak with a read-to-write ratio of 10:1, and 10 TB of usable storage per node.

• Number of items: 100 TB / 10 KB = 10 billion items.
• Stored bytes with replication: 100 TB x 3 = 300 TB.
• Nodes for storage: 300 TB / 10 TB per node = 30 nodes. Storing at 50% disk utilization to leave room for compaction and growth doubles that to about 60 nodes.
• Request rate per node: 1,000,000 / 60 ≈ 17,000 requests per second per node (each request touches several replicas for writes, so actual internal load is higher). This is within what a node with SSDs and an LSM-tree engine sustains.
• Network: 1,000,000 requests x 10 KB = 10 GB/s across the cluster, or about 170 MB/s per node before replication traffic. Replication multiplies write traffic: with a 10:1 read-to-write ratio, writes are 1/11 of requests, about 91,000 writes per second. 91,000 x 10 KB x 2 extra replicas ≈ 1.8 GB/s of replication traffic across the cluster.

See back-of-envelope estimation.

What each number implies:

• 10 billion items, 30 million per shard-node: each node holds about 10B x 3 / 60 = 500 million keys. A Bloom filter at about 10 bits per key (roughly 1% false positives) costs 500M x 10 bits = 625 MB of memory per node, which is affordable and avoids most needless SSTable reads.
• 17,000 requests/s per node: workable with SSDs and an LSM engine, but the 10 KB values mean about 170 MB/s of network per node, so 10 Gbps NICs are adequate and 1 Gbps is not.
• 1.8 GB/s replication traffic: the write path, not the read path, sets the network budget; this is a reason to allow lower W for non-critical data.
• 60 nodes at 50% disk: any node failure means re-streaming about 5 TB (see failure modes).

API design

put(key, value, context?, consistency?) -> ok | error
get(key, consistency?)                  -> { values: [value...], context } | not_found
delete(key, context?)                   -> ok

• context carries version metadata (see below). A get returns it and the following put passes it back, so the store can tell which version the client updated.
• consistency is an optional level (for example ONE, QUORUM, ALL), which sets how many replicas must respond.
• A get can return several values when concurrent writes produced conflicting versions, and the client or a resolution rule merges them.

Clients connect to any node. That node becomes the coordinator for the request.

Data model

The unit of storage is a (key, value, version) record. The store has no schema, no joins, and no multi-key transactions. Operations on one key are the unit of atomicity. This restriction is what makes horizontal partitioning simple: every operation involves a single key and therefore a single set of replicas. For richer data models and access patterns, see key-value and document stores.

On disk, each node stores its records in an LSM-tree engine (see the deep dive), which fits a write-heavy, key-lookup workload.

High-level design

flowchart TB
    C[Client] --> N1["Node A: coordinator"]
    N1 --> N2["Node B: replica"]
    N1 --> N3["Node C: replica"]
    N1 --> N4["Node D: replica"]
    subgraph NI["Node internals"]
        API[Request handler] --> WAL[Commit log]
        API --> MT[Memtable]
        MT --> SS[SSTables on disk]
    end

Every node runs the same software (peer-to-peer). Node responsibilities:

• Partitioning. Each key maps to a position on a hash ring, which determines the owner node.
• Replication. Each key is stored on N nodes (here N = 3).
• Coordination. Any node can accept a request and forward it to the replicas, then combine their replies.
• Membership and failure detection. Nodes learn who is alive through gossip.
• Storage engine. Each node persists data locally.

Write and read flow with N = 3, W = 2, R = 2:

sequenceDiagram
    participant C as Client
    participant Co as Coordinator
    participant R1 as Replica 1
    participant R2 as Replica 2
    participant R3 as Replica 3
    C->>Co: put(k, v)
    par send to all replicas
        Co->>R1: write k, v, version
    and
        Co->>R2: write k, v, version
    and
        Co->>R3: write k, v, version
    end
    R1-->>Co: ack
    R2-->>Co: ack
    Co-->>C: ok after W = 2 acks
    R3-->>Co: ack later

Deep dive

Partitioning and replication with consistent hashing

Hashing key mod N would move nearly every key when N changes. Consistent hashing avoids that: nodes and keys are hashed onto a ring, and a key belongs to the first node clockwise from its position. Adding or removing a node moves only the keys in adjacent ranges, about 1/N of the data. Each physical node owns many virtual nodes (positions on the ring), which evens out load and lets a new node take small slices from many existing nodes. See consistent hashing.

For replication, the key is stored on its owner and on the next N-1 distinct physical nodes clockwise. That list is the key's preference list. Skipping virtual nodes that belong to a physical node already in the list ensures replicas sit on different machines. Placing them in different racks or availability zones further limits correlated failures.

import java.nio.charset.StandardCharsets;
import java.util.ArrayList;
import java.util.List;
import java.util.NavigableMap;
import java.util.TreeMap;

public class Main {
    static long fnv1a(String s) {
        long h = 0xcbf29ce484222325L;
        for (byte b : s.getBytes(StandardCharsets.UTF_8)) {
            h ^= (b & 0xffL);
            h *= 0x100000001b3L; // long multiplication wraps on overflow
        }
        return h;
    }

    static final class Ring {
        // ring position (compared as unsigned 64-bit) -> physical node
        private final NavigableMap<Long, String> points = new TreeMap<>(Long::compareUnsigned);

        Ring(List<String> nodes, int vnodes) {
            for (String n : nodes) {
                for (int v = 0; v < vnodes; v++) {
                    points.put(fnv1a(n + "#" + v), n);
                }
            }
        }

        // First n distinct physical nodes clockwise from the key's position.
        List<String> preferenceList(String key, int n) {
            long start = fnv1a(key);
            List<String> out = new ArrayList<>();
            var clockwise = new ArrayList<String>(points.tailMap(start, true).values());
            clockwise.addAll(points.headMap(start, false).values());
            for (String node : clockwise) {
                if (!out.contains(node)) {
                    out.add(node);
                    if (out.size() == n) {
                        break;
                    }
                }
            }
            return out;
        }
    }

    public static void main(String[] args) {
        var ring = new Ring(List.of("A", "B", "C", "D", "E"), 16);
        for (String key : List.of("user:1", "user:2", "cart:9")) {
            System.out.println(key + " -> " + ring.preferenceList(key, 3));
        }
    }
}

user:1 -> [C, A, B]
user:2 -> [C, A, B]
cart:9 -> [B, A, D]

The nodes and virtual node count here are illustrative. The output shows each key's three distinct replicas in ring order.

Tunable quorums

With N replicas, a write waits for W acknowledgments and a read waits for R replies. If R + W > N, every read set overlaps every write set in at least one replica, so a read sees the latest acknowledged write (absent concurrent writes and sloppy quorums). See quorums and replication protocols.

The coordinator sends each operation to all N replicas and waits for the first W or R responses, so the slowest replica does not set latency.

Sloppy quorum and hinted handoff. If a preference-list node is unreachable, the coordinator writes to the next healthy node on the ring, which stores the write with a hint naming the intended owner. When the owner recovers, the hint holder forwards the data and deletes its copy. This keeps writes available during failures, but it weakens the overlap guarantee, because the write may sit on a node outside the preference list that a later read will not contact.

Concrete settings and rationale: use 128 to 256 virtual nodes per machine (with few vnodes, one node's removal dumps its load on one or two neighbors; with many, on all of them, at the cost of more ring metadata and more range overlap). Choose N = 3, W = 2, R = 2 as default because a single slow or dead replica does not affect either path, and the p99 depends on the second-fastest replica rather than the slowest. W = 3 makes p99 track the slowest of three. Sloppy quorum is for write availability, so pair it with a bounded hint window (for example 3 hours); beyond it, rely on anti-entropy repair.

Versioning and conflict resolution

Because any replica can accept a write during a partition, two clients can update the same key on different replicas. The store must detect this and resolve it.

• Last write wins (LWW). Each write carries a timestamp, and the highest timestamp survives. It is simple and needs no client involvement, but clock skew can discard a write that actually happened later, and concurrent updates are silently lost. Cassandra uses this approach.
• Vector clocks. Each value carries a vector clock, a list of (node, counter) pairs. Comparing two clocks tells whether one version descends from the other (the older one can be discarded) or whether they are concurrent (both are kept as siblings). The client then merges siblings on the next write, using the context returned by get. Dynamo used this approach. Vector clocks grow with the number of coordinators and are usually truncated. See time, clocks and ordering.
• CRDTs. Data types designed so concurrent updates merge deterministically (counters, sets). No conflict resolution by clients is needed for those types.

Deletes are written as tombstones, records marking a key as deleted. A tombstone must be retained long enough to propagate to all replicas. Otherwise a replica that missed the delete would send the old value back during repair.

Failure detection and repair

• Gossip. Each node periodically exchanges its membership and heartbeat table with a few random peers. Information about joins and failures spreads to all nodes in a logarithmic number of rounds, without a central registry. A failure detector such as the phi-accrual detector turns heartbeat gaps into a suspicion level. See gossip protocol and failure detection.
• Read repair. When a read contacts several replicas and finds stale versions, the coordinator writes the newest version back to the stale replicas.
• Anti-entropy with Merkle trees. Replicas periodically compare a hash tree over each key range. Nodes compare root hashes, and descend only into subtrees that differ, so they transfer just the differing keys. This repairs data that is never read and so never triggers read repair.
• Temporary versus permanent failure. A short outage is handled by hinted handoff. Permanent loss of a node requires re-replicating its ranges from the remaining replicas, which is a streaming operation coordinated by the ring change.

Single-node storage engine

Each node persists data with a log-structured merge tree (LSM tree), which converts random writes into sequential ones. See storage engines: B-tree vs LSM-tree.

Write path:

1. Append the record to a commit log (write-ahead log) on disk. This makes the write durable after a crash.
2. Insert it into the memtable, an in-memory sorted structure.
3. Acknowledge the write.
4. When the memtable reaches a size limit, flush it to disk as an immutable, sorted SSTable file, and discard the corresponding commit log segment.

Read path:

1. Check the memtable.
2. Check SSTables from newest to oldest. Each SSTable has a Bloom filter that says with certainty when a key is absent, so most SSTables are skipped without a disk read. See probabilistic data structures.
3. Within a matching SSTable, use a sparse index to locate the block and read it.
4. Return the newest version found.

Typical parameters: the memtable flushes at roughly 64 to 256 MB; the commit log is fsynced either on every write (durable, higher latency) or in periodic batches (for example every 10 ms, so a crash can lose up to that window on a single node, which replication covers). Compaction strategy is a trade-off: size-tiered compaction gives lower write amplification (roughly 4 to 10x) but higher read amplification and up to 2x temporary space; leveled compaction gives few SSTables per read (good for read-heavy loads) at 10x or more write amplification. Tombstones should be kept longer than the maximum time a replica can be down or unrepaired (for example 10 days).

Compaction merges SSTables in the background, dropping overwritten values and expired tombstones, so read cost and disk use stay bounded. Compaction consumes disk and I/O bandwidth, which is why the estimate above leaves spare disk capacity.

Failure modes

• One node down (N = 3, W = R = 2). Reads and writes for its ranges continue on the other two replicas; hinted handoff covers writes until it returns.
• Two nodes down. With 256 vnodes per node, nearly every pair of nodes shares some range, so any two simultaneous failures make QUORUM unavailable for some keys, even though 58 nodes are healthy. Degrade by allowing ONE for reads that tolerate staleness, and place replicas in different racks and zones so one rack failure removes at most one replica.
• Rebuild window. Re-streaming 5 TB at a throttled 100 MB/s takes about 14 hours; during that time the affected ranges have one fewer replica. Faster rebuilds, or smaller nodes, shrink the exposure.
• Partition. The side with fewer than the required replicas rejects strict-quorum requests; conflicting siblings appear after healing and are merged as described above.
• Disk full during compaction. Keep disk under about 50 to 70% utilization; alert well before.
• At 10x scale (600 nodes): gossip still converges in O(log N) rounds, but node failures occur about 10x as often (at a 5% annual node failure rate, about 30 per year versus 3), so repair, streaming, and hinted handoff are constant background activity and must be automated and throttled. Add rack-aware placement and a cap on concurrent streams per node.

Bottlenecks and scaling

• Adding nodes. A new node claims virtual node positions and streams the ranges it now owns from current owners. Streaming is rate limited so it does not starve foreground traffic.
• Hot keys. A single key lives on one preference list, so a very popular key overloads three nodes. Mitigations are a cache in front of the store, client-side caching, or splitting the key into several sub-keys (for example, appending a random suffix) and combining on read.
• Hot partitions and uneven load. More virtual nodes per machine reduce imbalance. Heterogeneous machines receive virtual nodes in proportion to capacity.
• Coordinator overhead. Clients that know the ring topology can send requests directly to a replica in the preference list, avoiding an extra hop.
• Compaction and repair load. Schedule them off-peak and throttle them. Repair traffic can significantly increase cluster load.
• Large values. Values well above the limit belong in object storage, with the key-value store holding a reference.
• Cross-datacenter replication. Configure replicas per datacenter and use local quorums (a quorum among replicas in the coordinator's own datacenter) so requests do not wait on the wide-area network.

Trade-offs and follow-up questions

• Why not strong consistency? Achieving it requires a consensus protocol or single-leader replication per partition, and writes then block during leader failure or partitions. Systems such as Spanner and etcd take that route and give up some availability. See consensus: Paxos and Raft.
• How does the store behave in a partition? Each side that can reach enough replicas for its chosen R and W continues. With W = 1 both sides accept writes and the versions conflict later. With W = 2 of 3 and a strict quorum, only the side holding at least two replicas accepts writes (with a sloppy quorum, the minority side can also accept writes through hinted handoff).
• LWW or vector clocks? LWW is simpler and loses concurrent updates. Vector clocks preserve them at the cost of client-side merging and metadata growth.
• How to provide transactions across keys? This design does not. Add a layer with two-phase commit or use a store with a consensus-based transaction protocol. See distributed transactions: 2PC and 3PC.
• How to choose the number of replicas? N = 3 tolerates one failure with a quorum of 2 and survives the loss of one availability zone when replicas are placed in three zones.
• What if a client needs read-your-writes? R + W > N gives it when replicas are reached by strict quorum; otherwise route a session to the same coordinator or pass the write's version as the context and require reads at least that new.
• What if the coordinator crashes after W acks but before replying? The write is durable on W replicas; the client times out and retries the same put with the same version, which is idempotent.
• How would you implement an atomic counter? A plain read-modify-write loses increments. Use a CRDT counter (per-node increments merged by max and sum) or a conditional put with the version context.
• What if a key's value grows to 50 MB? Chunk it into sub-keys plus a manifest, or store the blob in object storage and keep a pointer.
• How would you add a new datacenter? Add it to the replication configuration, stream existing data, then use local quorums for reads and writes.
• How would range scans work? Hash partitioning scatters adjacent keys. Range queries need order-preserving partitioning (as in Bigtable) or a compound key with a partition key and a sorted clustering key (as in Cassandra). See sharding and partitioning.