SDE Path

Replication & Sharding: leader/follower, failover, shard keys, resharding pain

13 min read

Once a single database hits its limit, you have two scaling directions: replication (copies for read scale) and sharding (split data across databases for write scale). Both add complexity.

Replication: Leader-Follower

How it works:

Write traffic → Leader (primary database)
                  ↓ (binary log / WAL)
              Replica 1 (read-only follower)
              Replica 2 (read-only follower)
              Replica 3 (read-only follower)

Read traffic can hit replicas, spreading load.

Mechanism:

  1. Write hits leader: INSERT, UPDATE committed.
  2. Leader records change in binary log.
  3. Replicas continuously replay log (asynchronous).
  4. Reads hit replicas, reducing load on leader.

Pros:

  • Scales reads horizontally: 1 leader + 100 replicas = 100x read scale.
  • High availability: if leader dies, promote a replica (with downtime).
  • Geographic distribution: replicas in different regions for local reads.

Cons:

  • Eventual consistency: replicas lag leader (by milliseconds to seconds).
  • Replication lag problems: if you read your own write immediately, you might get stale data.
  • Write scale doesn't improve: all writes still hit one leader.
  • Failover complexity: promoting replica requires DNS update, app reconnection, handling in-flight transactions.

Common lag issue:

User updates name: UPDATE users SET name="Alice" WHERE id=123;
  → Leader commits.
  → Replica still has old name for 100 ms.
User immediately reads: SELECT name FROM users WHERE id=123;
  → If read hits replica, returns old name.
  → Frontend shows wrong name!

Solution: Stick writes and reads on the leader, or use sticky sessions (read from leader for current user).


Sharding: Splitting Data

How it works:

Shard Key: user_id

Shard 0: user_id % 16 == 0 → [1, 17, 33, ...]
Shard 1: user_id % 16 == 1 → [2, 18, 34, ...]
...
Shard 15: user_id % 16 == 15 → [16, 32, 48, ...]

Each shard is an independent database (possibly with replicas).

Pros:

  • Scales writes: split load across 16 databases, ~16x write throughput.
  • Each shard is smaller: queries are faster.
  • Distribute load: can use different hardware per shard.

Cons:

  • Shard key is critical: if you choose wrong (e.g., shard by country, and one country has 90% traffic), skew is permanent.
  • Distributed queries fail: to find "all users named Alice," you must query all shards (slow).
  • Transactions across shards are expensive: UPDATE user 123 AND update their account—if they're on different shards, distributed transaction (slow, risky).
  • Resharding is painful: changing from 16 shards to 32 requires recomputing hash for every row, moving data.

Resharding: The Pain

Scenario: Started with 4 shards. Now you have 2TB per shard and need 8 shards.

The process:

  1. Create 8 new databases (empty).
  2. For each row in old shards: compute new_shard = hash(key) % 8; copy to new DB.
  3. Update routing logic: query 8 shards instead of 4.
  4. Decommission old 4 shards.

Reality:

  • Backfill 2TB: can take hours or days (network I/O).
  • Data verification: ensure all rows moved correctly (spot checks, checksums).
  • Downtime risk: if backfill fails halfway, you have inconsistent state.
  • Read traffic: split between old and new shards during migration (double-check both).

Modern tools (e.g., Vitess) automate resharding, but it's still complex.


Shard Key Selection

Good shard key:

  • High cardinality: user_id (billions of distinct values) > country (200 distinct values).
  • Even distribution: hash-based so user 1 and user 2 go to different shards, not skewed.
  • Immutable: user_id is stable; email can change.

Bad shard key:

  • Low cardinality: shard by country → if one country is 90% traffic, one shard is overloaded.
  • Mutable: shard by email → if email changes, row is on wrong shard.
  • Temporal: shard by date → old shards are cold, new shards are hot.

Combining Replication and Sharding

Most large systems use both:

Shard 0 (user_id 0-4M):
  → Leader (writes)
  → Replica 1 (reads)
  → Replica 2 (reads, different region)

Shard 1 (user_id 4M-8M):
  → Leader (writes)
  → Replica 1 (reads)
  → Replica 2 (reads, different region)

... (16 shards total)

Each shard has replication for availability. Sharding handles write scale.


In the interview

  1. Replication for reads, sharding for writes. Clear distinction.

  2. Mention the shard key: "I'd shard by user_id, not country, because user_id has high cardinality and even distribution."

  3. Discuss replication lag: "Reads might be eventually consistent. For read-your-write, I'd hit the leader or use a cache."

  4. Acknowledge resharding pain: "Adding shards later is complex—choosing the shard key initially is critical."

  5. Prefer replication initially. "Start with 1 leader + replicas for scale. Only shard when leader write throughput is the bottleneck."

Check yourself

5 questions — graded when you submit.

Question 1

In leader-follower replication, why does replication lag cause the read-your-own-write problem?

Question 2

You shard by email address, but emails can change. Why is this a bad shard key?

Question 3

You have 4 shards and need to scale to 8. Why is resharding painful?

Question 4

Sharding by country (200 distinct values) might cause problems. Why?

Question 5

When should you introduce sharding in a growing system?

Finished reading?

Sign in to save your progress across lessons.