Amazon Dynamo: Highly Available Key-Value Store

Ideas you will meet

Replication

Leader/follower, multi-leader and leaderless replication.

CAP Theorem

Consistency, availability and partition tolerance trade-offs.

Consistency Models

Linearizable, causal and eventual consistency explained.

What we already know about this

Our own lessons on these ideas. Read them here before you start, so it makes sense.

• System Design Foundations · 11 minReplication and shardingTwo ways to spread data over many machines: copy it for safety and read speed, or split it for capacity.Read hereClose

  Replication: keep copies

  Replication keeps the same data on more than one machine. Reasons:

  • Availability: if one machine dies, another has the data.
  • Read capacity: reads can go to any copy.
  • Locality: place a copy near users in another region.

  The most common arrangement is leader and followers. All writes go to the leader, which sends the changes to followers. Reads can use followers.

  100%

  Figure SDA-FND-05-1 · Leader and follower replication · © forcsguy.com

  Many teams hide the split between the primary and the replicas behind a small router, so the application does not need to know:

  100%

  Figure SDA-FND-05-2 · Read and write splitting through a router · © forcsguy.com

  Synchronous or asynchronous

  • Synchronous: the leader waits for followers to confirm before saying "done". Safer, but slower, and a down follower can block writes.
  • Asynchronous: the leader confirms immediately and followers catch up. Fast, but a follower can be behind, and a leader crash can lose the most recent writes.

  Most systems use asynchronous replication for most followers, sometimes with one synchronous follower as a safety net.

  Replication lag

  Because followers can lag, a user may write something and then read an older value from a follower. Fixes include reading the user's own recent writes from the leader, or tracking how fresh each replica is. This is the first taste of a bigger idea, consistency, covered in lesson 7.

  When the leader fails

  A follower must be promoted (failover). That needs failure detection and agreement on who the new leader is, so that two machines do not both believe they are the leader (a split brain). This is why teams rely on tested tooling or managed databases for failover rather than hand-written scripts.

  Read-after-write consistency

  With asynchronous replicas, a user can save something and then not see it: the write went to the primary, the next read hit a replica that is a few seconds behind. Three common fixes:

  • Send reads that must be fresh (the user's own profile, an order just placed) to the primary.
  • For a short window after a user writes, route that user's reads to the primary.
  • Let the router check whether a replica has caught up to the user's last write; if not, read from the primary or wait.

  A database proxy in front of the cluster can do this routing so application code stays simple, at the cost of one more hop and one more component that must itself be highly available.

  Before: one database

  100%

  After: primary with read replicas

  100%

  What changed

  • Reads are spread across copies
  • A copy can be promoted if the primary fails
  • Replicas may lag slightly behind the primary

  Sharding: split the data

  Replication does not help when the data or the writes no longer fit on one machine. Sharding (partitioning) divides the data so each shard holds a part. Each shard is usually replicated too.

  100%

  Figure SDA-FND-05-3 · Sharding by key range · © forcsguy.com

  Choosing a shard key

  The shard key decides which shard owns a row. A good key:

  • Spreads data evenly, so no shard is much bigger than the others.
  • Spreads traffic evenly, so no shard is a hot spot.
  • Matches your queries, so most queries touch one shard.

  Common schemes:

  Scheme	How	Watch out for
  Range	key ranges per shard	hot spots, e.g. newest dates
  Hash	hash(key) mod N	adding shards remaps most keys
  Directory	a lookup table says where each key lives	the directory is another thing to run and scale
  Consistent hashing	keys and nodes placed on a ring	needs virtual nodes to balance well

  Consistent hashing in one picture

  Place both servers and keys on a ring. A key belongs to the first server clockwise from it. Adding a server only takes keys from its neighbour; the rest do not move.

  100%

  Figure SDA-FND-05-4 · Consistent hashing ring · © forcsguy.com

  Horizontal and vertical partitioning

  Sharding is horizontal partitioning: the rows of a table are split across machines, each shard holding the same columns for a different set of rows. Vertical partitioning splits by columns or by feature instead: rarely used or large columns (a profile photo, a long description) move to a separate table or store, or whole groups of tables go to different databases (users on one, orders on another). It is simpler than sharding and keeps frequently read data small and fast, but it does not help when a single table's rows outgrow one machine.

  100%

  Figure SDA-FND-05-5 · Splitting a table by rows and by columns · © forcsguy.com

  What sharding costs you

  • Cross-shard queries and joins become slow or need extra work.
  • Transactions across shards are hard (lesson 8 touches on how systems cope).
  • Rebalancing when a shard gets too big is operational work.
  • More moving parts to monitor.

  A worked example: sizing replicas and shards

  Replicas for reads. A social app has 20,000 reads and 1,000 writes per second. One database node can do about 5,000 simple operations per second. A primary alone is hopeless. With writes of 1,000 per second on the primary, you have capacity left for about 4,000 reads, so you need at least four replicas to serve 20,000 reads (16,000 on replicas plus 4,000 on the primary) with no headroom, so six in practice. Adding replicas scales reads almost linearly. It does nothing for the 1,000 writes, which still all go to one primary.

  Shards for writes. If writes grow to 12,000 per second, no single primary can take them. Split the data into shards, each with its own primary: with 4 shards each takes about 3,000 writes per second. Each shard then has its own replicas for reads. Your final count is shards x (1 primary + replicas), for example 4 x (1 + 2) = 12 database nodes.

  100%

  Figure SDA-FND-05-6 · Replicas scale reads, shards scale writes · © forcsguy.com

  Walking through a failover

  Replication is only useful if you know what happens when a primary dies. With asynchronous replication:

  1. The primary acknowledges writes 100 to 140, but the follower has only received up to 135 (replication lag).
  2. The primary crashes. Writes 136 to 140 exist only on the dead machine.
  3. A monitor notices missing heartbeats, usually after several seconds, and promotes the most up-to-date follower to primary.
  4. Clients are redirected. Writes 136 to 140 are lost (or recoverable only when the old primary returns).
  5. When the old primary comes back it must not accept writes as if nothing happened (that would create two primaries, "split brain"); it rejoins as a follower and discards or reconciles what the new primary does not have.

  With synchronous replication to at least one follower, step 2 loses nothing, but every write waits for that follower, adding latency and making writes fail if the follower is unavailable. Many systems choose a compromise: wait for one follower to confirm (semi-synchronous), accepting a little latency to avoid losing acknowledged writes.

  100%

  Figure SDA-FND-05-7 · What an asynchronous failover can lose · © forcsguy.com

  Choosing a shard key: three examples

  • User id for a messaging app. Almost every query is "this user's messages", so a hash of the user id keeps each user's data together and spreads users evenly. The pain: a query across all users (an admin search) must ask every shard.
  • Order date for an orders table. Range sharding by date makes "last month's orders" cheap, but all new writes land on the newest shard, a hot spot, while old shards sit idle. A better key adds a hash of the customer id.
  • Country for a multi-region product. Good for keeping data near users and meeting data-residency rules, but shards are uneven: one country may hold half your users.

  A good key has high cardinality (many distinct values), even load and matches your most common query. Changing it later means moving most of your data, so test it against a sample of real traffic first.

  Mistakes people make

  • Reading from a replica right after writing. The replica may not have the write yet, so users do not see their own change. Read your own writes from the primary for a short time, or route that user to the primary.
  • Sharding too early. Sharding multiplies operational work: backups, schema changes, cross-shard queries and transactions all get harder. First try indexes, caching, replicas and a bigger machine.
  • Counting on replicas as backups. A mistaken delete is replicated within milliseconds. Keep real backups and test restoring them.
  • Forgetting to rebalance. Shards grow unevenly. Plan how to split a shard, ideally with consistent hashing or many small virtual shards that you can move between machines.

  Try it yourself

  In the System Builder, open "The database goes down". Run it with no replicas and see reads and writes fail, then add replicas and a write queue and compare. For hashing itself, build the consistent hash ring lab and measure how many keys move when you add a node. To place replicas and a queue yourself, open the database-outage design on the free canvas and kill the database.

  More on this site

  Related reading

  • How Discord Stores Trillions of MessagesBo Ingram · DiscordBlogs
  • MongoDB: Shard KeysMongoDB · MongoDB Inc.Documentation
  • Uber Schemaless: A Scalable MySQL DatastoreUber Engineering · UberCase Studies
  • How Discord Stores Billions of MessagesDiscordCase Studies

  Show 6 moreShow fewer

  • Consistent Hashing and Random Trees (paper)Karger et al. · MITPapers
  • Amazon Dynamo: Highly Available Key-Value StoreWerner Vogels · AmazonCase Studies
  • Mitigating MySQL Replication Lag and Reducing Read Load at GithubGithub engineers · GithubCase Studies
  • MongoDB at Baidu: Multi-Tenant Cluster Storing 200+ Billion Documents across 160 ShardsMongodb engineers · MongodbCase Studies
  • Shopify EngineeringShopify engineers · ShopifyBlogs
  • Sharding MySQL at QuoraQuora engineers · QuoraCase Studies

  Open this lesson with its quiz and exercise

• System Design Foundations · 12 minConsistency, availability and the CAP theoremWhat goes wrong when data lives on several machines, and the vocabulary for choosing what to guarantee.Read hereClose

  The problem

  On one machine, data has one value. Spread it over several machines and new questions appear:

  • If I write to machine A, when can machine B see it?
  • If A and B cannot talk to each other, should they keep serving requests?
  • If two people change the same item at the same moment, who wins?

  Consistency models are the answers to the first question. The CAP theorem frames the second.

  Consistency: what a read may return

  From strongest to weakest, in plain words:

  • Strong (linearizable): once a write completes, every later read anywhere sees it. It behaves like one copy. Easiest to reason about, slowest and least available.
  • Read-your-writes: you always see your own updates, though others may lag.
  • Eventual: if writes stop, all copies will eventually agree. Reads may be stale in between. Fast and highly available.

  100%

  Figure SDA-FND-07-1 · Replica lag and eventual consistency · © forcsguy.com

  Stronger guarantees cost latency (replicas must coordinate) and availability (some must be reachable). Weaker ones are cheaper but push complexity to the application.

  The CAP theorem

  For a distributed data store, three properties are in tension:

  • Consistency (C): every read sees the latest write (in the strong sense above).
  • Availability (A): every request to a working node gets a response.
  • Partition tolerance (P): the system keeps working even when messages between nodes are lost or delayed.

  Networks do fail, so P is not optional. The real choice is what to do when a partition happens:

  100%

  Figure SDA-FND-07-2 · Choosing C or A during a partition · © forcsguy.com

  • CP systems (for example many coordination stores and strongly consistent databases) may reject requests rather than risk a wrong answer.
  • AP systems (for example many Dynamo-style stores) stay up and let replicas diverge briefly, then merge.

  A more useful lens: latency too

  Even without partitions, strong consistency costs time because replicas must agree. The PACELC idea adds this: if there is a Partition choose A or C, Else (normal operation) choose between Latency and Consistency. It explains why a store may feel slower when configured to be strongly consistent.

  100%

  Figure SDA-FND-07-3 · How many copies must confirm a write · © forcsguy.com

  Resolving conflicts

  If two replicas accept different writes, the system needs a rule:

  • Last write wins: simple, but silently drops one update.
  • Version vectors or merge logic: detect that the writes were concurrent, then merge or ask the user.
  • CRDTs: data types designed so concurrent updates always merge cleanly (used for counters and collaborative editing).

  Choosing per feature, not per system

  You rarely need one consistency level for the whole product. Use strong consistency where wrong data is costly (payments, inventory, seat booking) and eventual consistency where speed and uptime matter more (counts, feeds, recommendations). Many databases let you pick per query.

  A worked example: quorums with numbers

  A key is stored on N = 3 replicas. A write must be acknowledged by W of them, and a read asks R of them and takes the newest version.

  Setting	W	R	R + W > N?	What you get
  Fast reads	3	1	yes (4 > 3)	Strong reads, but one dead replica blocks writes
  Balanced	2	2	yes (4 > 3)	Strong reads; survives one replica down for both reads and writes
  Fast writes	1	3	yes (4 > 3)	Quick writes, slow reads, reads fail if any replica is down
  Very fast	1	1	no (2 < 3)	Fastest, but a read can miss the latest write: eventual consistency

  The rule R + W > N guarantees the set of replicas read overlaps the set that acknowledged the last write, so at least one has the newest value. Concretely, with W = 2, R = 2: a write of version 5 reaches replicas A and B; C is still at version 4. A read that asks B and C sees versions 5 and 4 and returns 5. Whichever two replicas you ask, at least one has version 5.

  100%

  Figure SDA-FND-07-4 · Why read and write quorums overlap · © forcsguy.com

  A story: the comment that disappeared

  A user posts a comment. The write goes to the primary and the page reloads, but the read is served by a replica that is 300 ms behind. The comment is missing, the user posts it again, and now there are duplicates.

  This is a read-your-writes violation, the most common user-visible consistency bug. Fixes, from cheapest to strongest:

  1. For a short time after a user writes (say 5 seconds), route that user's reads to the primary.
  2. Return the new item in the write response and show it on the page without re-reading.
  3. Track a version or log position from the write, and have the replica serve the read only when it has caught up to it.
  4. Make that data strongly consistent end to end (higher latency, fewer failure modes tolerated).

  Not every read deserves strong consistency. A like counter that is a few seconds old is fine; a balance after a transfer is not.

  What a network partition actually does

  Imagine a three-node store split into two nodes on one side and one node on the other.

  • A CP system (consistency over availability) lets only the side with a majority accept writes. The lone node refuses requests, which is unavailable but correct.
  • An AP system (availability over consistency) lets both sides accept writes. Everyone gets an answer, but the two sides now hold different versions, and after the partition heals the system must merge: last-write-wins (simple, can lose data and trusts clocks), version vectors that keep both and ask the application, or data types designed to merge (CRDTs such as counters and sets).

  100%

  Figure SDA-FND-07-5 · Behaviour during a partition · © forcsguy.com

  Mistakes people make

  • Reading CAP as "pick any two". Partitions happen whether you like it or not, so the real choice is what to give up during a partition. Without partitions you can have both.
  • Assuming "eventual" means "soon enough". It means "if writes stop, replicas converge", with no bound. Measure the real lag.
  • Last-write-wins on unsynchronised clocks. A node with a clock a second ahead silently overwrites newer data. Prefer logical versions.
  • One setting for the whole system. Choose per feature: strong for payments and inventory, eventual for counters, feeds and recommendations.
  • Forgetting the latency side. Even without partitions, stronger consistency costs round trips between replicas, a trade-off often called PACELC.

  Try it yourself

  Use the CAP theorem playground to partition a cluster and watch CP and AP behave differently, and the replication playground to see stale reads and what a failover loses. Then try the System Builder with replicas.

  More on this site

  Related reading

  • Amazon Dynamo: Highly Available Key-Value StoreWerner Vogels · AmazonCase Studies
  • Please stop calling databases CP or APMartin Kleppmann · martin.kleppmann.comBlogs
  • CAP Twelve Years Later: How the Rules Have ChangedEric Brewer · InfoQBlogs
  • Martin Kleppmann's BlogMartin Kleppmann · martin.kleppmann.comBlogs

  Show 6 moreShow fewer

  • Martin Kleppmann: Distributed Systems LecturesMartin Kleppmann · University of CambridgeYouTube
  • Amazon S3 strong consistencyAmazon Web ServicesDocumentation
  • Designing Data-Intensive ApplicationsMartin Kleppmann · O'ReillyBooks
  • Read Consistency with Database Replicas at ShopifyShopify engineers · ShopifyCase Studies
  • All Things DistributedWerner Vogels · AmazonBlogs
  • Aurora BlogAurora engineers · AuroraBlogs

  Open this lesson with its quiz and exercise

• Classic System Designs · 14 minDesign a key-value storeA Dynamo-style distributed store: partitioning, replication, quorums, conflict handling and how reads and writes reach disk.Read hereClose

  The problem

  A key-value store supports put(key, value) and get(key). A distributed one must also:

  • Store far more than one machine can hold.
  • Keep serving when machines or networks fail.
  • Have low latency for both reads and writes.
  • Let you tune consistency against availability.

  We use a Dynamo-style design, which is the model behind several widely used stores.

  Partition the data

  Spread keys across nodes with consistent hashing (previous lesson). Virtual nodes keep the load even and let you add capacity gradually. Any node can act as the coordinator for a request, so there is no single leader to bottleneck on.

  Replicate each key

  Copy every key to N nodes: the owner and the next N-1 distinct nodes clockwise on the ring. Spread replicas across racks or zones so one failure does not take out all copies.

  100%

  Figure SDA-CSD-03-1 · Replicating each key to N nodes · © forcsguy.com

  Tunable consistency with quorums

  Choose three numbers:

  • N: number of replicas.
  • W: replicas that must acknowledge a write.
  • R: replicas that must answer a read.

  Setting	Behaviour
  W + R > N	Reads overlap the latest write: strong-ish consistency
  W = 1, R = 1	Fastest and most available, but reads can be stale
  W = N, R = 1	Fast reads, writes fail if any replica is down

  You pick the trade-off per use case, which is what makes this design flexible.

  A quorum write, step by step

  100%

  Figure SDA-CSD-03-2 · Coordinator and a write quorum (N=3, W=2) · © forcsguy.com

  When replicas disagree

  Replicas can hold different versions, for example after a partition.

  • Versioning: attach a version (a vector clock) to each value so the system can tell whether one version supersedes another or they were concurrent.
  • Reconciliation: for concurrent versions, either the application merges them (shopping cart: union the items) or the system uses a simple rule such as last-write-wins, which can silently discard data.

  Two writers at once: conflicts and vector clocks

  100%

  Figure SDA-CSD-03-3 · Concurrent writes create two versions · © forcsguy.com

  100%

  Figure SDA-CSD-03-4 · Vector clocks across writes · © forcsguy.com

  A clock A is an ancestor of B if every counter in A is less than or equal to the matching counter in B. If each has some counter larger than the other's, the versions are siblings and need merging.

  Failure handling

  • Detection: nodes exchange gossip messages about who is alive, so there is no central monitor to fail.
  • Temporary failure: if a replica is down, another node accepts the write and stores a hint, handing it over when the replica returns (hinted handoff).
  • Permanent repair: periodically compare replicas using Merkle trees (hashes of key ranges) and copy only what differs (anti-entropy).
  • Data centre failure: replicate across regions for the keys that need it.

  Noticing failures: gossip

  100%

  Figure SDA-CSD-03-5 · Gossip failure detection · © forcsguy.com

  Staying available through a short outage

  100%

  Figure SDA-CSD-03-6 · Sloppy quorum and hinted handoff · © forcsguy.com

  Repairing replicas for good: Merkle trees

  100%

  Figure SDA-CSD-03-7 · Comparing replicas with Merkle trees · © forcsguy.com

  More failure-handling detail

  • Do not trust one witness. One node saying "that server is down" is not enough; require agreement from several independent sources (gossip gives this naturally: a node is marked down only after others also see its heartbeat counter stall).
  • Sloppy quorum. A strict quorum can block when the usual replicas are down. Instead, take the first healthy servers on the ring for the write or read, and let a stand-in hold the data with a "hint". When the real replica returns, the stand-in hands the data back (hinted handoff). You gain availability and give up the strict overlap guarantee for a while.
  • Replica placement. When walking the ring for N replicas, skip virtual nodes that belong to a server already chosen, and spread replicas across data centers, because nodes in one site often fail together.
  • Bigger machines. Give a more powerful server more virtual nodes so it takes a proportionally larger share.
  • Conflicting versions. Vector clocks grow as more servers touch a key; cap their length and accept that, rarely, old entries dropped make some conflicts harder to untangle.

  Inside one node: how writes and reads hit disk

  100%

  Figure SDA-CSD-03-8 · Write and read path on one node · © forcsguy.com

  Write path

  1. Append to a commit log (sequential, fast, crash-safe).
  2. Apply to an in-memory sorted structure (the memtable).
  3. When the memtable is full, flush it to an immutable sorted file on disk; background compaction merges files.

  Read path

  1. Check the memtable.
  2. If missing, check the files, using a Bloom filter per file to skip files that definitely do not contain the key.
  3. Merge results, newest wins.

  This log-structured design turns random writes into sequential ones, which is why such stores handle huge write volumes.

  One node, many jobs

  100%

  Figure SDA-CSD-03-9 · What every node does · © forcsguy.com

  Summary of the design

  Concern	Technique
  Scale out	Consistent hashing with virtual nodes
  Durability and availability	N replicas across zones
  Consistency dial	Quorums (W, R)
  Conflicts	Vector clocks and merge logic
  Failure detection	Gossip
  Temporary and permanent repair	Hinted handoff, Merkle-tree anti-entropy
  Fast writes	Commit log, memtable, sorted files

  Worked numbers: how many nodes?

  Target: 1 billion keys averaging 1 KB, with 3 replicas, serving 100,000 operations per second.

  • Raw data: 1 B x 1 KB = 1 TB. With three replicas, 3 TB of storage across the cluster, before overhead for indexes and compaction (plan for about 2x headroom on disk for compaction, so around 6 TB).
  • Per-node capacity: if one node comfortably holds 1 TB of data and serves about 10,000 operations per second, 12 to 15 nodes cover both storage (3 TB / 1 TB, with room) and throughput (100,000 / 10,000 x 3 for replication on writes and headroom).
  • With a 20-node ring each node owns about 5% of the keys, and losing one node moves its 5% (about 150 GB of replicas) to its neighbours: repair takes minutes to hours, which is why you keep spare capacity.

  When a replica is down: hinted handoff

  With W = 2 of N = 3, a write must reach two replicas. Suppose node C is down.

  1. The coordinator sends the write to A, B and C; A and B acknowledge, so the write succeeds.
  2. For C, a hint ("this write was meant for C") is stored on a healthy node.
  3. When C returns, the holder of the hint replays the write to C.

  If the node is down for too long the hints are dropped, and a slower anti-entropy repair takes over.

  100%

  Figure SDA-CSD-03-10 · A write while one replica is down · © forcsguy.com

  Anti-entropy with Merkle trees

  Replicas can still drift apart (a hint was lost, a disk was corrupted). Comparing every key between two replicas is far too expensive. A Merkle tree summarises a key range with hashes: each leaf is a hash of a small bucket of keys, and each parent is a hash of its children. Two replicas compare root hashes; if equal, the whole range matches. If not, they compare the children and descend only into the branches that differ, until they find the few buckets that actually disagree and exchange just those keys.

  100%

  Figure SDA-CSD-03-11 · Finding differences without comparing every key · © forcsguy.com

  Mistakes people make

  • Hot keys. One extremely popular key overloads its replicas whatever the cluster size. Cache it in front or split it into several keys.
  • Large values. Multi-megabyte values hurt latency and replication; store big blobs in object storage and keep a pointer.
  • Expecting joins and secondary queries. A key-value store answers "get by key". If you need other queries, add an index structure or use another database.
  • Resolving conflicts with wall-clock time alone. Skewed clocks lose writes; prefer version vectors or let the application merge.
  • Forgetting tombstones. Deletes leave markers that must live long enough to reach all replicas, and be compacted away later. Skipping this resurrects deleted data.
  • Never running repair. Without regular anti-entropy, rarely read data can silently diverge.

  Try it yourself

  Build the consistent hash ring lab to see how few keys move when a node joins, the Bitcask lab for one node's storage engine, and the write-ahead log lab for crash recovery. Then read the consistency and CAP lesson for the quorum rule used above.

  More on this site

  Related reading

  • Amazon Dynamo: Highly Available Key-Value StoreWerner Vogels · AmazonCase Studies
  • The Google File System (paper)Ghemawat, Gobioff and Leung · GooglePapers
  • The Log-Structured Merge-Tree (paper)O'Neil et al. · UMass BostonPapers
  • Time, Clocks, and the Ordering of Events in a Distributed System (paper)Leslie Lamport · Microsoft ResearchPapers

  Show 6 moreShow fewer

  • ZippyDB: Facebook's key-value storeMeta Engineering · MetaCase Studies
  • Inside the Magic PocketDropboxCase Studies
  • Spanner: Google's Globally Distributed Database (paper)Corbett et al. · GooglePapers
  • Read Consistency with Database Replicas at ShopifyShopify engineers · ShopifyCase Studies
  • Designing Large-Scale SystemsSomethingsimilar engineers · SomethingsimilarBlogs
  • Reddit Inc Blog HomepageRedditblog engineers · RedditblogBlogs

  Open this lesson with its quiz and exercise