Sharding

When one notebook won't close

Replicas fixed reads. Writes still all land in one place, and eventually that one place is the wall: one machine's disk, one machine's memory, one machine's write throughput. No index and no photocopy helps, because the problem is volume.

So split the notebook. Customers A through M in the first notebook, N through Z in the second. Two notebooks, each half the size, each taking half the writes. That is sharding, and each piece is a shard.

It works. It is also the most expensive decision in this course, because from this moment on the shop can no longer answer 'how many orders did we take today?' by reading one book. It has to ask every notebook and add up the answers.

Imagine it like this: Splitting the customer ledger into A-M and N-Z. Twice the capacity, but nobody can answer a question about all customers without visiting both books.

Words you just learned

shard: - one slice of the data, living on its own machine
shard key: - the column that decides which shard a row belongs to

Choosing the key is the whole game

The shard key decides everything: whether load spreads evenly, whether your common queries touch one shard or all of them, and whether you can ever change your mind cheaply.

Split alphabetically and you will discover that far more customers' names start with S than with X. Split by signup date and every new write lands on the newest shard while the others idle. Both are hot partitions - shards taking far more than their share while the rest of your expensive cluster does nothing.

Hashing the key spreads load beautifully, because a hash of sequential ids scatters them. The price is that ranges are destroyed: 'every order from last Tuesday' now means asking every shard, because those rows were deliberately scattered.

shard_keys.py

python

customers = ["ana", "arjun", "sam", "sana", "sofia", "steve", "xavier"]

def by_letter(name):          # range shard: A-M -> 0, N-Z -> 1
    return 0 if name[0] < "n" else 1

def by_hash(name):            # hash shard: scatter deliberately
    return sum(ord(c) for c in name) % 2

for shard_fn in (by_letter, by_hash):
    counts = [0, 0]
    for c in customers:
        counts[shard_fn(c)] += 1
    print(shard_fn.__name__, counts)

What it prints

by_letter [2, 5]
by_hash   [4, 3]

Same seven customers. The range split put five of seven on one machine while the other loafed; the hash split is nearly even. But only the range split can answer 'all customers starting with S' without asking everyone.

The celebrity problem

Sharding by user_id looks perfectly even until one user has forty million followers. That single row's traffic lands on one shard, and no amount of hashing helps, because it is one key. Real systems special-case these rows - caching them hard, or splitting that user's data by a second dimension.

Words you just learned

hot partition: - a shard receiving far more traffic than its share
range sharding: - splitting by value ranges - good for range queries, prone to hotspots
hash sharding: - splitting by a hash - spreads evenly, destroys range queries

Everything gets harder once, and forever

Sharding is not a setting you turn on. It changes what your database is able to do, permanently.

A JOIN across two shards is not a JOIN any more - it's two queries and application code stitching results together. A transaction spanning shards needs distributed coordination, which is slow and can fail halfway. ORDER BY created_at LIMIT 10 across eight shards means fetching ten from each, merging eighty, and discarding seventy.

And then there is resharding. Going from 8 shards to 16 changes where every row belongs, which means moving data while the site is live and writes keep arriving. This is why teams pick a shard count they will not outgrow, and why consistent hashing exists - it moves only the keys that must move rather than reshuffling everything.

Shard last, not first

In order: index properly, add a cache, add read replicas, buy a bigger machine. Modern hardware takes a single Postgres a very long way. Shard only when writes - not reads - are the wall, because every option above is reversible and this one is not.

Remember these

Cross-shard JOINs become application code
Cross-shard transactions become slow and failure-prone
Sorting and pagination must gather from every shard, then merge
Resharding is a live data migration - plan the key so you rarely do it

Words you just learned

resharding: - changing the number of shards, which relocates data while the system runs
consistent hashing: - a mapping where adding a shard moves only a small fraction of keys