S074 · Sharding & Partitioning Strategies

Module M08: Distributed Systems · Session 74 of 130 · Track: SYS 🧩

What you'll be able to do after this session

• Explain Sharding & Partitioning Strategies to a friend in one minute, out loud, no notes.
• Recognise it when you see it in real code / systems / papers.
• Complete the hands-on exercise at the end without looking up help.

Prerequisites

If you can't answer these in 30 seconds each, redo the linked session first:

• S071 — Replication — Leader/Follower, Multi-Leader, Leaderless

Pre-read (skim before the session)

• 🎥 Intuition (5–15 min) · Database Sharding Explained (Hussein Nasser) — 15 min, why and how sharding works in plain English.
• 🎥 Deep dive (30–60 min) · How Discord Stores Billions of Messages (Fireship + Discord blog) — real-world sharding walkthrough at social-network scale.
• 🎥 Hands-on demo (10–20 min) · Consistent Hashing Explained (ByteByteGo) — 12-minute animated walkthrough of the algorithm behind most modern shard maps.
• 📖 Canonical article · MongoDB Docs — Sharding — the canonical hands-on reference; skim "Shard Keys" and "Zone Sharding."
• 📖 Book chapter / tutorial · DDIA Ch.6 — Partitioning (O'Reilly) — Kleppmann's chapter is the definitive treatment.
• 📖 Engineering blog · Discord — How Discord Stores Trillions of Messages — the migration to ScyllaDB and how they picked a shard key.

(a) Intuition · 5 min

The one-paragraph mental model. Why this concept exists, what problem it solves, and the analogy that makes it stick.

In one sentence: Split your data before it splits you.

Imagine your company has grown from 10 employees to 10 million customers, and the single filing cabinet where you kept everyone's records is now the size of a warehouse — and every lookup takes 20 minutes because the janitor has to walk to the far end. Sharding is the moment you say "one cabinet is not enough; let's split customers A–F into cabinet 1, G–M into cabinet 2, and so on." Now every clerk works on a smaller cabinet, in parallel.

Sharding (a.k.a. horizontal partitioning) exists because a single machine has finite CPU, RAM, disk, and IOPS. Once your dataset or write rate outgrows one box, you must split it — vertical scaling ("just buy a bigger box") stops at the biggest AWS instance and costs 10× per doubling. Sharding scales linearly and cheaply, but adds a permanent tax: every query now has to figure out which shard holds the data.

The three shard-key strategies you'll see: range-based (customers A–F on shard 1 — easy range scans, but hot shards if data is skewed), hash-based (hash(customer_id) mod N — perfectly even, but no range scans), and directory-based / consistent hashing (a lookup table or ring — flexible, expensive to maintain).

Gotcha to carry forever: the shard key is the most important schema decision you'll ever make, and it is nearly impossible to change later. Pick a key that (a) spreads writes evenly, (b) matches your most common query pattern, and (c) has enough cardinality to grow. Wrong shard key = you rewrite the app and migrate petabytes.

(b) Visual walkthrough · 15 min

Diagrams, worked examples, step-by-step visuals.

Worked example — why hash-mod-N is a resharding trap.

You start with 4 shards and hash by user_id:

Now you add a fifth shard (mod 5). Recompute:

About 80% of data has to physically move — a nightmare at TB scale. This is why consistent hashing exists: adding a node moves only ~1/N of the data, because the ring is only rebalanced at the new node's segment.

Real example — Discord's shard key. Discord shards messages by (channel_id, bucket) where bucket = message_id / (10 days worth of snowflakes). Why?

1. Users overwhelmingly read recent messages in a specific channel → shard key aligns with query pattern.
2. Old buckets become read-mostly → they can be moved to cheaper storage.
3. Cardinality is enormous (millions of channels × many buckets) → no single hot shard.

Compare to a naive user_id shard key: one user sending 10 000 messages/day would make one shard a hotspot forever.

Hot shards and how to spot them. Monitor per-shard QPS and disk usage. If one shard is 5× the median, you have a hot shard — usually because a "celebrity" key gets all the traffic. Mitigation: split the hot key into sub-shards (user_123:0, user_123:1, …).

(c) Hands-on · 20 min

Code you type and run. Not pseudo-code — real, runnable, copy-pasteable.

Implement consistent hashing from scratch in ~40 lines of Python and confirm the "1/N moves on rebalance" property.

# consistent_hash.py — python3, no deps
import hashlib, bisect, collections
 
class ConsistentHash:
    def __init__(self, nodes=None, vnodes=100):
        self.vnodes = vnodes            # virtual nodes per real node
        self.ring = []                  # sorted list of hashes
        self.hash_to_node = {}
        for n in (nodes or []):
            self.add(n)
 
    def _hash(self, key):
        return int(hashlib.md5(key.encode()).hexdigest(), 16)
 
    def add(self, node):
        for i in range(self.vnodes):
            h = self._hash(f"{node}#{i}")
            bisect.insort(self.ring, h)
            self.hash_to_node[h] = node
 
    def remove(self, node):
        for i in range(self.vnodes):
            h = self._hash(f"{node}#{i}")
            self.ring.remove(h)
            del self.hash_to_node[h]
 
    def get(self, key):
        h = self._hash(key)
        idx = bisect.bisect(self.ring, h) % len(self.ring)
        return self.hash_to_node[self.ring[idx]]
 
def assign(keys, nodes, vnodes=100):
    ch = ConsistentHash(nodes, vnodes)
    return {k: ch.get(k) for k in keys}
 
if __name__ == "__main__":
    keys = [f"user_{i}" for i in range(100_000)]
 
    # 1. baseline: 4 shards
    before = assign(keys, ["A", "B", "C", "D"])
    # distribution
    dist = collections.Counter(before.values())
    print("4 nodes distribution:", dist)
 
    # 2. add a 5th shard
    after = assign(keys, ["A", "B", "C", "D", "E"])
    moved = sum(1 for k in keys if before[k] != after[k])
    print(f"Moved after adding node E: {moved}/{len(keys)} = {moved/len(keys):.1%}")
 
    # 3. compare with naive mod-N
    def mod_assign(keys, n):
        return {k: chr(ord('A') + (hash(k) % n)) for k in keys}
    before_mod = mod_assign(keys, 4)
    after_mod  = mod_assign(keys, 5)
    moved_mod  = sum(1 for k in keys if before_mod[k] != after_mod[k])
    print(f"Moved (mod-N naive): {moved_mod}/{len(keys)} = {moved_mod/len(keys):.1%}")

Observe (checklist):

1. Distribution across 4 nodes should be within ±5% (100 vnodes is enough for evenness).
2. Adding node E moves about 20% of keys — that's 1/N = 1/5.
3. Naive mod rebalancing moves ~80% of keys — the disaster.
4. Try vnodes=1: distribution becomes lumpy (up to 2× skew). Increasing vnodes trades memory for evenness.
5. Removing a node moves only that node's slice; the rest stay put.

Try this modification: simulate a "hot key" by adding keys += ["celebrity"] * 50_000. Observe that one node gets slammed. Then implement key splitting: hash celebrity:0, celebrity:1, …, celebrity:9 and confirm the traffic spreads.

(d) Production reality · 10 min

How this shows up in real systems, gotchas, war stories, common bugs.

Sharding is where most teams learn the phrase "you can't ship your way out of a bad schema decision."

Instagram's Postgres shard-by-user-id. Their canonical 2012 blog post described 4096 logical shards mapped onto a smaller number of physical Postgres instances — the logical/physical decoupling meant they could rebalance without changing the app. This is the pattern almost every modern SaaS copies.

Figma's 2023 sharding migration. They ran one giant Postgres for years, hit a wall, and spent a year moving to horizontally sharded Postgres — using CoLo groups (co-locating a team's data on one shard so joins work). Their public writeup is the best modern case study.

Common war stories:

• The celebrity problem. One user (Cristiano Ronaldo on Instagram, a mega-guild in a game) causes 100× the write rate of a normal user. Their shard becomes a hotspot; the DB dies. Fix: shard by a composite key that spreads that user's data, or use "buckets" (Discord's approach).
• Cross-shard joins are painful. Once sharded, a JOIN across two users on different shards is either a distributed query (slow, complex) or done in the app layer (also slow). Denormalise ruthlessly.
• Cross-shard transactions. A "transfer money from user A to user B" transaction, if A and B are on different shards, needs 2-phase commit or Saga. Every serious sharding system either (a) forbids this or (b) makes it painful — you'll want to design your shard key to avoid it.
• Rebalancing is a project, not a command. Adding a new shard = weeks of live data migration with dual writes, backfill, cutover, and cleanup. Vitess and Citus automate this well; homegrown solutions rarely do.
• Backup/restore complexity. A point-in-time restore that's consistent across 128 shards requires coordinated snapshots — not something the standard tooling handles.

Rule of thumb: shard as late as you can, then shard using a battle-tested framework (Vitess for MySQL, Citus for Postgres, MongoDB sharding, or a purpose-built distributed DB like CockroachDB / Yugabyte). Rolling your own is a rite of passage very few teams enjoy.

(e) Quiz + exercise · 10 min

1. What is the difference between range-based and hash-based sharding, and when would you use each?
2. Why is choosing the wrong shard key so expensive to fix later?
3. Explain the "hot shard" problem and one concrete mitigation.
4. In one sentence: what advantage does consistent hashing have over hash mod N?
5. Why are cross-shard transactions difficult, and what techniques do teams use to avoid needing them?

Stretch (connects to S071 · Replication): Sharding and replication are orthogonal — you can (and should) do both. Sketch the storage layout for a 6-node cluster that has 3 shards, each replicated 2×, using a leader/follower model. How many node failures can this survive?

Explain-out-loud test

If you can't teach these 3 things to a friend without notes, redo the session:

1. What is Sharding & Partitioning Strategies? (one sentence, no jargon)
2. When would you use it? (one concrete example)
3. When would you NOT use it? (one anti-pattern)

What comes next

• Next session (S075): Message Queues — SQS, RabbitMQ, Kafka as Queue
• Previous (S073): Consensus — Paxos & Raft Intuition
• Hub: The 6-Month Learning Plan

Part of a 130-session evergreen learning series. Session structure: (a) intuition · (b) visual walkthrough · (c) hands-on · (d) production reality · (e) quiz + exercise. Duration: 60 minutes.