Database Sharding Explained: Shard Keys, Hot Shards, Resharding

Sharding is the answer to a specific problem: one database can no longer keep up with the writes. Read replicas do not help with that, and a bigger machine only buys time. Splitting the data itself does, at a price. In a system design interview, saying "we shard it" is easy. What the interviewer is listening for is the shard key you would pick, what happens to queries that do not carry it, and what you would do when one shard runs hot.

The problem it solves

A bigger machine, caching and read replicas all leave one thing untouched: every write still goes to one primary, and its disk, memory and CPU set the ceiling.

Sharding, also called horizontal partitioning, splits one dataset across several independent databases. Each shard owns a slice of the rows and serves reads and writes for that slice. Three shards can absorb roughly three times the write traffic, as long as the traffic spreads evenly across them. That condition is where every difficulty of sharding comes from.

The intuition

Everything hangs on the shard key, the column that decides which shard owns a row. A router, either a library in the application or a proxy, maps each key to a shard. Two common mappings:

• Range sharding: each shard owns a range of keys, such as birth years 1946 to 1964. Range queries stay on one shard, but a popular range makes that shard hot.
• Hash sharding: hash the key and map hash buckets onto shards. Load spreads evenly, but a range of keys is scattered across every shard.

A good shard key appears in nearly every query, so the router can send each query to one shard, and it spreads load evenly. user_id is a common choice for user-centric data because most requests are "this user's things".

The cost shows up in everything that crosses shards. A query without the shard key has to be sent to every shard and the results merged in the router, which is called scatter-gather. Joins across shards, unique constraints on other columns and transactions that touch several shards get hard or disappear. And changing the number of shards moves data: with hash mod N, going from 3 shards to 4 moves about three quarters of all keys, which is why consistent hashing exists.

Watch it run

The animation starts with one database whose writes no longer fit on one box. Replication cannot help, because every write still lands on one primary. It picks a shard key that appears in nearly every query, user_id, and maps its hash buckets onto three physical databases: one dataset, three owners. A query carrying the key hashes to bucket 1 and touches shard B alone. The next user hashes to bucket 2, and shard C answers that one. Each shard owns its slice and serves its own writes, so write capacity finally scales out. Then a search by surname: nothing says which slice holds the answer, so every shard is asked, and their results have to be merged, with any ordering redone in the router. Worse, the key can be uneven: in the animation, shard B holds the busiest years and does 62% of the work. The last two frames list the bill: cross-shard joins, unique constraints and transactions, and resharding a live system, which is a migration project.

The code

A toy model, not a real database: one list of rows per shard, and a router that hashes the key with zlib.crc32, since Python's hash of a string changes between runs. get carries the key; find scatters to every shard and sorts the merged result:

import zlib

def bucket(key, n):
    return zlib.crc32(str(key).encode()) % n        # stable hash, same on every run

class ToyShardedDB:
    """Toy model of hash sharding behind a router, not a real database."""
    def __init__(self, n_shards, shard_key):
        self.shards = [[] for _ in range(n_shards)]
        self.shard_key = shard_key

    def insert(self, row):
        self.shards[bucket(row[self.shard_key], len(self.shards))].append(row)

    def get(self, value):
        """A query that carries the shard key: one shard answers."""
        s = self.shards[bucket(value, len(self.shards))]
        return [r for r in s if r[self.shard_key] == value], 1

    def find(self, predicate, order_by):
        """No shard key: scatter to every shard, gather, and re-sort in the router."""
        hits = []
        for rows in self.shards:
            hits += [r for r in rows if predicate(r)]
        return sorted(hits, key=order_by), len(self.shards)

Six users on three shards. A lookup by user_id touches one shard; a lookup by surname touches all three:

db = ToyShardedDB(3, "user_id")
people = [(71, "Okafor", 1958), (72, "Lind", 1961), (73, "Okafor", 1990),
          (74, "Silva", 1957), (75, "Lind", 2001), (76, "Moreau", 1960)]
for uid, surname, year in people:
    db.insert({"user_id": uid, "surname": surname, "year": year})

print([len(s) for s in db.shards])                  # [2, 2, 2]
rows, touched = db.get(73)
print(rows, touched)
# [{'user_id': 73, 'surname': 'Okafor', 'year': 1990}] 1
rows, touched = db.find(lambda r: r["surname"] == "Lind", order_by=lambda r: r["year"])
print([r["user_id"] for r in rows], touched)        # [72, 75] 3

Hot shards and resharding, measured. Range-sharding 10,000 people by birth year, when half were born in a 19-year boom, puts half the rows on one shard; hashing the user id spreads them evenly. Then the keys that move under hash mod N when one shard is added:

import random
from collections import Counter

random.seed(22)
years = [random.choice(range(1946, 1965)) if random.random() < 0.5
         else random.choice(range(1965, 2006)) for _ in range(10000)]
by_range = Counter(0 if y < 1965 else 1 if y < 1985 else 2 for y in years)
by_hash = Counter(bucket(uid, 3) for uid in range(10000))
print([round(100 * by_range[s] / 10000) for s in range(3)])   # [51, 24, 25]
print([round(100 * by_hash[s] / 10000) for s in range(3)])    # [33, 34, 33]

keys = range(100000)
for old, new in ((3, 4), (10, 11)):
    moved = sum(bucket(k, old) != bucket(k, new) for k in keys)
    print(old, "->", new, f"{100 * moved / len(keys):.0f}% of keys move")
# 3 -> 4 75% of keys move
# 10 -> 11 91% of keys move

The router checked against an unsharded table on 1,000 random datasets with 1 to 6 shards: every row stored once, on its key's shard, and both kinds of query matching a plain filter and sort:

random.seed(22)
ok = True
for _ in range(1000):
    n = random.randint(1, 6)
    t = ToyShardedDB(n, "user_id")
    table = [{"user_id": random.randint(0, 50), "surname": random.choice("ABCDE"),
              "year": random.randint(1940, 2010)} for _ in range(random.randint(0, 40))]
    for row in table:
        t.insert(row)
    # every row lives on exactly one shard, the one its key hashes to
    ok &= sorted(map(repr, sum(t.shards, []))) == sorted(map(repr, table))
    ok &= all(bucket(r["user_id"], n) == i for i, s in enumerate(t.shards) for r in s)
    uid = random.randint(0, 50)
    rows, touched = t.get(uid)
    ok &= touched == 1 and sorted(map(repr, rows)) == sorted(repr(r) for r in table if r["user_id"] == uid)
    name = random.choice("ABCDE")
    rows, touched = t.find(lambda r: r["surname"] == name, order_by=lambda r: (r["year"], r["user_id"]))
    want = sorted((r for r in table if r["surname"] == name), key=lambda r: (r["year"], r["user_id"]))
    ok &= touched == n and [(r["year"], r["user_id"]) for r in rows] == [(r["year"], r["user_id"]) for r in want]
print(ok)                                           # True

The complexity

With S shards:

• Key lookup: one shard, so the cost is the same as on an unsharded database of 1/S the size.
• Query without the key: S shard queries plus a merge in the router, as slow as the slowest shard.
• Write capacity: up to S times one database if writes spread evenly; a skewed key caps it at the hottest shard.
• Resharding with hash mod N: about 1 - 1/(N + 1) of the keys move when one shard is added: 75% going from 3 to 4, 91% from 10 to 11.

Where it goes wrong

• Sharding too early. Replicas, caching, indexes and a bigger machine are all cheaper. Shard when writes, or the size of the data, no longer fit.
• A monotonic key with range sharding. Keys such as timestamps or auto-increment ids send every new write to the last shard.
• A key that most queries do not carry. Every query becomes scatter-gather, and adding shards makes each one more expensive.
• Celebrity keys. One very popular value can take a large share of the traffic even under hashing; DynamoDB write sharding splits such a key with a suffix.
• Assuming cross-shard transactions. Keep one request's writes on one shard where possible.

When it shows up in interviews

Sharding appears in almost every "design X at scale" question, usually after the interviewer pushes back with "your database cannot handle this write rate." It connects to the CAP theorem, since each shard is usually also replicated, and to cases such as designing a URL shortener, where the short code is a natural shard key. Name the key, the queries it serves and does not, and the plan for adding shards.

How to say it in an interview

"Replicas scale reads, but every write still goes to one primary, so when writes outgrow it I shard. I pick a key that is in nearly every query and spreads load, here user_id, and hash it; a router sends each query to the shard that owns the key. Queries without the key become scatter-gather, so I keep them rare or serve them from a separate index. I watch for hot keys, keep one user's data on one shard to avoid cross-shard transactions, and use consistent hashing so that adding a shard moves only a small fraction of the data."