Designing Data-Intensive Applications Part 2: Distributed Data

In Part 1, we examined storage engines operating on a single machine. But as data volume grows, or as latency demands require placing data close to users worldwide, a single machine becomes a bottleneck and a single point of failure.

We must distribute data across multiple machines connected over a network.

There are two primary dimensions of data distribution:

1. Replication: Keeping copies of the same data on multiple nodes for redundancy, fault tolerance, and read scalability.
2. Partitioning (Sharding): Splitting a massive dataset into smaller, independent subsets across nodes to scale beyond single-machine storage capacity.

In Part 2 of our Designing Data-Intensive Applications masterclass series, we unpack the algorithms, trade-offs, and failure modes of distributed data.

1. Replication: Three Architecture Patterns

Replication means storing copies of the exact same data on multiple physical machines (replicas). Every write to the database must be processed by replicas; otherwise, the copies diverge.

There are three major architectural patterns for handling writes in replicated databases:

1. SINGLE-LEADER REPLICATION
Client Write ──> [ LEADER ] ──> Follower 1 (Async/Sync)
                           └──> Follower 2 (Async)

2. MULTI-LEADER REPLICATION
Client Write ──> [ LEADER Datacenter US ] <─── Replication ───> [ LEADER Datacenter EU ] <── Client Write

3. LEADERLESS REPLICATION (Dynamo-style)
Client Write ──> Parallel Writes to [ Node A, Node B, Node C ] (Quorum W + R > N)

Pattern 1: Single-Leader Replication (Master-Slave)

Single-leader replication is the standard model used by PostgreSQL, MySQL, Oracle, and MongoDB.

1. One replica is designated as the leader (master/primary). All client write requests (INSERT, UPDATE, DELETE) must be sent directly to the leader.
2. The leader writes data to its local storage, and streams the data change as a replication log or change stream to all followers (slaves/read replicas).
3. Each follower takes the log from the leader and updates its local copy of the database in the exact same sequence.
4. Read queries (SELECT) can be served by the leader or any of the followers.

Client (Write) ─────> [ LEADER ] ── (Write Log) ──> [ FOLLOWER 1 ] (Sync - OK)
                         │
                         └─── (Write Log) ──────> [ FOLLOWER 2 ] (Async)

Synchronous vs Asynchronous Replication

• Synchronous: The leader waits until Follower 1 confirms it received the write before returning success to the client. Guarantee: Follower is 100% up to date. Downside: If Follower 1 fails or network drops, the leader blocks all writes!
• Asynchronous: The leader returns success to the client immediately after writing locally, and sends logs to followers in the background. Guarantee: Blazing fast writes. Downside: If the leader crashes, un-replicated writes are permanently lost!

Industry Standard: Most production single-leader databases use Semi-Synchronous Replication (1 synchronous follower + N asynchronous followers).

Handling Node Outages: Failover

If the leader crashes, a follower must be promoted to the new leader via automatic Failover:

1. Detect leader failure via heartbeats (e.g. no response for 30 seconds).
2. Choose a new leader via consensus (the follower with the most up-to-date replication log).
3. Reconfigure the system to route writes to the new leader.

Danger (Split-Brain): If two nodes both believe they are the legitimate leader (due to a temporary network partition), both will accept writes. When the network recovers, data is hopelessly corrupted unless split-brain detection automatically shoots the old leader in the head (STONITH).

Pattern 2: Multi-Leader Replication (Active-Active)

Single-leader replication breaks down when you operate across multiple datacenters worldwide. Sending all writes from Europe to a single leader in North America creates unacceptable network latency.

In Multi-Leader Replication, each datacenter has its own leader node that accepts local writes. Leaders replicate their changes to each other asynchronously.

   [ US DATACENTER ]                          [ EU DATACENTER ]
 Client ──> [ Leader US ] <─── Async Stream ───> [ Leader EU ] <── Client
                 │                                   │
                 ▼                                   ▼
            [ Follower ]                        [ Follower ]

The Ultimate Challenge: Conflict Resolution

If User A in New York changes a document title to “Project Alpha” at 10:00:00.001, and User B in London changes the same document title to “Project Beta” at 10:00:00.002, both leaders accept their local write. When the replication streams cross, a Write Conflict occurs!

Conflict Resolution Strategies:

1. Conflict Avoidance: Ensure all writes for a specific record are routed to the same datacenter (e.g. route user based on home region).
2. Last Write Wins (LWW): Assign a timestamp to each write and keep only the latest write. Warning: Clock drift across servers makes LWW drop valid writes silently!
3. Merge Values / CRDTs: Use Conflict-Free Replicated Data Types (like Observed-Remove Sets) to automatically merge concurrent updates mathematically.

Pattern 3: Leaderless Replication (Dynamo-Style)

Pioneered by Amazon’s Dynamo paper and implemented in Apache Cassandra, Amazon DynamoDB, and Riak, leaderless replication abandons the concept of a leader entirely.

Clients send write and read requests directly to multiple parallel nodes.

Quorum Reads and Writes (W + R > N)

If there are N total replicas in the cluster:

• A write must be acknowledged by at least W nodes to be considered successful.
• A read must query at least R nodes to retrieve data.

Example Quorum Configuration:

• N = 3 (3 total replicas)
• W = 2 (Write succeeds if 2 nodes acknowledge)
• R = 2 (Read queries 2 nodes and takes the newest timestamp)
• Since 2 + 2 = 4 > 3, quorum holds!

How Leaderless Systems Heal Stale Replicas:

1. Read Repair: When a client queries R = 2 nodes and detects that Node A has version 5 while Node B has version 4, the client writes version 5 back to Node B in the background.
2. Anti-Entropy Process: A background process constantly uses Merkle Trees (hash trees) to compare dataset differences between replicas and sync missing keys.

2. Replication Lag Anomalies

When applications read from asynchronous followers, they experience eventual consistency. If you stop writing, all followers will eventually catch up with the leader.

However, the delay between a write on the leader and its arrival at a follower (the replication lag) can range from milliseconds to minutes. This lag introduces severe user-facing anomalies.

Anomaly 1: Reading Your Own Writes (Read-After-Write Consistency)

Imagine a user posts a comment on a forum, reloads the page, and the comment disappears! Why? The write went to the leader, but the reload read query was routed to a lagging follower that hasn’t received the write yet.

User Action ──> [ WRITE ] ──> Leader (OK)
                                │ (Replication Lag: 5 seconds)
User Reload ──> [ READ ]  ──> Follower 2 (Stale Data! Comment missing!)

Solution: Read-After-Write Consistency Guarantees

• If a user modifies data, always read that user’s own data from the leader for a set time (e.g. 1 minute after write).
• Monitor follower replication lag, and prevent read routing to followers that are more than 2 seconds behind the leader.

Anomaly 2: Monotonic Reads (Time Moving Backwards)

A user reads a thread. Query 1 hits Follower 1 (which is 1 second behind), returning 5 comments. The user refreshes; Query 2 hits Follower 2 (which is 10 seconds behind), returning only 3 comments! To the user, time appears to move backward.

Read 1 (t=0s) ──> Follower 1 (Lag: 1s)  ──> Returns 5 Comments
Read 2 (t=1s) ──> Follower 2 (Lag: 10s) ──> Returns 3 Comments (Time WENT BACKWARDS!)

Solution: Monotonic Reads Guarantee

Ensure that each user always reads from the same replica (e.g., hash the user ID to select a specific follower node). If that replica fails, fall back to another.

Anomaly 3: Consistent Prefix Reads (Causal Violations)

Consider a dialogue between two people:

• Alice: “How far into the future can you see?”
• Bob: “About 5 seconds.”

If a follower receives Bob’s answer before Alice’s question due to replication lag across partitions, an observer reading from that follower sees:

• Bob: “About 5 seconds.”
• Alice: “How far into the future can you see?”

Solution: Consistent Prefix Reads

Guarantee that if a sequence of writes happens in a certain order, anyone reading those writes will see them in the same order. This requires routing causally dependent writes to the same partition.

3. Partitioning (Sharding): Scaling Big Data

Replication alone is insufficient when your total dataset exceeds the disk capacity of a single machine. To scale further, we must divide the database into partitions (shards in MongoDB/Elasticsearch, vnodes in Cassandra).

Every partition acts as a mini-database of its own.

MASSIVE 3TB DATABASE
  ├── Partition 1 (Keys A - G) ──> Node 1 (1TB)
  ├── Partition 2 (Keys H - P) ──> Node 2 (1TB)
  └── Partition 3 (Keys Q - Z) ──> Node 3 (1TB)

Partitioning Strategies

The goal of partitioning is to spread data and query load evenly across nodes. If partitioning is uneven, some partitions will have far more data or queries than others—this is called a hot spot or skew.

Strategy 1: Partitioning by Key Range

Assign a continuous range of keys (e.g., A-C, D-F) to each partition, similar to volumes of a paper encyclopedia.

• Pros: Range queries are extremely efficient (e.g., fetch all users whose names start with ‘B’).
• Cons: Severe hot spots if keys are sequential (e.g., timestamp keys 2025-06-09-00:01 mean today’s writes all smash into a single partition!).

Strategy 2: Partitioning by Hash of Key

Apply a cryptographic hash function (like MD5 or MurmurHash3) to the key, and assign ranges of hashes to partitions.

• Pros: Spreads keys uniformly, eliminating hot spots.
• Cons: Range queries are ruined! Keys that were adjacent (e.g., timestamps) are scattered across all partitions on disk.

Consistent Hashing & Rebalancing

When your application grows and you add 5 new nodes to the cluster, some data must move from existing nodes to the new nodes (Rebalancing).

Why hash(key) mod N is a Disaster:

If you partition using hash(key) % N (where N is the number of nodes), changing N from 10 to 11 forces almost 100% of all keys in the entire database to move to new nodes! This triggers a catastrophic network storm during rebalancing.

Solution: Fixed Partitions & Consistent Hashing

Create far more partitions than nodes (e.g. 1,000 fixed partitions for 10 nodes). Each node is assigned ~100 partitions. When a new 11th node joins, it simply steals ~9 partitions from each of the existing 10 nodes without re-hashing any keys!

[ Node 1 ] ──> Holds Partitions 1..100
[ Node 2 ] ──> Holds Partitions 101..200
Add Node 3 ──> Reassign Partitions 67..100 & 167..200 to Node 3! (Zero Key Re-hashing!)

Secondary Indexes in Partitioned Databases

Primary key lookups (SELECT * FROM users WHERE user_id = 123) route directly to the exact target partition. But what if you query by a secondary index (SELECT * FROM users WHERE country = 'Canada')?

There are two approaches to secondary indexes in sharded systems:

1. Document-Partitioned Index (Local Index)

Each partition maintains its own local secondary index for the data stored strictly inside that partition.

Partition 1 ──> Stores Local Index [Canada -> User 1, User 5]
Partition 2 ──> Stores Local Index [Canada -> User 12, User 19]

Query: "WHERE country = 'Canada'" ──> Scatter-Gather Query!
Must query ALL partitions simultaneously and combine results!

2. Term-Partitioned Index (Global Index)

The secondary index itself is partitioned globally across nodes by the index term value.

Node A ──> Stores Global Index for [Canada -> User 1, User 5, User 12, User 19]
Node B ──> Stores Global Index for [USA -> User 2, User 8, User 22]

Query: "WHERE country = 'Canada'" ──> Route directly to Node A! Fast Reads!
Drawback: Writes must update multiple remote index partitions asynchronously.

Summary of Part 2

In Part 2, we mastered the core mechanics of distributed data:

1. Single-Leader Replication offers straightforward consistency but suffers from single-point-of-failure writes.
2. Multi-Leader Replication suits multi-datacenter setups but introduces complex write conflict resolution challenges.
3. Leaderless Replication relies on Quorum math (W + R > N) to guarantee read consistency.
4. Replication Lag creates anomalies like loss of Read-After-Write consistency, non-monotonic reads, and causal ordering violations.
5. Partitioning scales storage capacity beyond single nodes; using Consistent Hashing prevents catastrophic network storms during node rebalancing.

Up next: Designing Data-Intensive Applications Part 3: Transactions & Consistency (ACID, MVCC, & Distributed Consensus).