HT
How Things Work
System Online

Sharding & Replication

How a database grows past one machine — partitioning data across shards by a key, replicating each shard for read scale and failover, and the consistency trade-offs that follow (CAP, replication lag).

How It Works

When a single database can't keep up, you scale it in two directions. Sharding partitions rows across many machines so each holds a slice of the data and the write load. Replication copies each shard so reads can fan out to replicas and a failed primary can fail over. Together they unlock scale — at the price of distributed-systems consistency trade-offs.

1
Partition the Data (Sharding)

When one database can't hold all the data or handle all the writes, split it horizontally into shards — each shard owns a disjoint subset of rows. A shard key (e.g. user_id) plus a routing function (hash or range) decides which shard owns each record.

2
Route Every Request

Reads and writes first compute the shard key, then route to the single shard that owns that key. Queries that include the shard key are fast (one shard); queries that don't (scatter-gather across all shards) are slow — so the shard key choice shapes performance.

3
Replicate for Reads & Durability

Within a shard, the primary takes all writes and streams them to replicas. Replicas serve read traffic (scaling reads) and stand by for failover. Replication is usually asynchronous, so replicas lag the primary by a small window.

4
CAP & Consistency

During the replication lag, a read from a replica can return stale data — eventual consistency. CAP says that under a network partition you must choose: stay Consistent (reject reads/writes) or stay Available (serve possibly-stale data). Most systems tune this per operation.

Key Concepts

🔑Shard Key

The field used to route a record to a shard. A good key spreads load evenly and keeps related data co-located.

📋Replication

Copying a shard's data to replicas for read scaling and failover. The primary takes writes; replicas follow.

⏱ïļReplication Lag

The delay before a write on the primary appears on replicas. Reads during this window are stale (eventual consistency).

⭕Consistent Hashing

A ring-based scheme where adding/removing a shard remaps only ~1/N of keys instead of all of them.

⚖ïļCAP Theorem

Under a network partition a distributed store can be Consistent or Available, not both. The C-vs-A choice defines its behavior.

ðŸ”ĨHot Shard

A single shard receiving disproportionate traffic because of a skewed shard key (e.g. one celebrity user).

Routing keys & replica reads
tsx
1// Routing a key to a shard.
2//
3// ❌ Modulo sharding — simple, but adding a shard remaps EVERY key
4function shardOf(key, shardCount) {
5 return hash(key) % shardCount; // 3→4 shards = mass reshuffle
6}
7
8// ✅ Consistent hashing — adding a node only moves ~1/N of keys
9// Place nodes and keys on a hash ring; a key belongs to the
10// next node clockwise. Add a node → only its arc is remapped.
11function shardOf(key, ring) {
12 const h = hash(key);
13 return ring.find(node => node.position >= h) ?? ring[0];
14}
15
16// Reads can be served by replicas (scales reads, but they may
17// be slightly stale). Writes always go to the primary.
18const user = await replica.query("SELECT * FROM users WHERE id=?", [id]);
19await primary.exec("UPDATE users SET name=? WHERE id=?", [name, id]);
ðŸ’Ą
Why This Matters

Sharding and replication are how every large data store (Cosmos DB, DynamoDB, Cassandra, Vitess, Citus) scales. The choices — shard key, sync vs async replication, where reads are served — determine throughput, latency, and what consistency anomalies your users will hit. It's the heart of distributed-data interview questions.

Common Pitfalls

⚠A skewed shard key creates hot shards that adding capacity can't fix.
⚠Modulo sharding: adding one shard remaps almost every key. Use consistent hashing.
⚠Cross-shard queries and transactions are slow and complex (scatter-gather, 2PC) — design to avoid them.
⚠Forgetting replication lag: replica reads can return stale data, breaking 'read your own write'.
⚠Async replication means a primary crash can lose the last un-replicated writes — know your durability guarantees.
Real-World Use Cases

1The Celebrity Hot Shard

Scenario

A social app shards posts by user_id. Most shards are fine, but the shard holding a few celebrity accounts is constantly overloaded while others sit idle.

Problem

Sharding by user_id assumes traffic is evenly distributed across users. Celebrities break that assumption — their shard becomes a hot spot, and you can't relieve it just by adding shards because all their data is on one.

Solution

Use a composite or higher-cardinality shard key (e.g. hash of post_id rather than user_id) so a single user's posts spread across shards, and cache hot reads aggressively at the edge. Some systems give hot tenants dedicated shards.

ðŸ’Ą

Takeaway: Sharding only helps if the shard key spreads load evenly. Skewed access patterns create hot shards that more shards alone won't fix — the key choice is everything.

2Reading Your Own Write — and Not Seeing It

Scenario

A user updates their profile, the write succeeds on the primary, but the very next page load (served by a replica) still shows the old name.

Problem

Reads were routed to replicas for scale, but asynchronous replication hadn't caught up yet. The user experienced the system as 'broken' even though it was just eventually consistent.

Solution

Apply read-your-writes consistency: route a user's reads to the primary (or a caught-up replica) for a short window after they write, or pin them to the primary via a session token. Use replicas freely for other users' data.

ðŸ’Ą

Takeaway: Eventual consistency is fine for most reads but jarring when users don't see their own changes. Read-your-writes routing closes that gap without giving up replica read-scaling everywhere else.

Join the discussion

Comments

?

Loading comments...

Cookie preferences

We use essential cookies for sign-in and preferences. With your permission, we also use basic analytics to improve lessons.