Sharding & Partition Keys

By the end of this lesson you will be able to calculate how many shards a system needs, choose between range, hash and directory partitioning, pick a partition key that spreads load evenly, and spot the hot-shard and cross-shard problems before they reach production.

When one database is not enough

Replication copies the same data to many machines, which helps reads. But every replica still holds all the data and every write still goes to one leader. When the data is too big for one machine, or writes are too many for one leader, you must split the data itself.

Sharding (horizontal partitioning) splits the rows of a table across several databases. Each piece is a shard, and each shard holds only part of the data. The column that decides which shard a row goes to is the partition key (or shard key).

Think of an exam centre with 6,000 students. One hall cannot seat them all, so students are split across ten halls by roll number range. Each hall has its own invigilator. Finding a student is easy if you know their roll number. Finding "all students named Kabir" means checking every hall. That trade-off, fast lookups by the key and slow lookups by anything else, is the core of sharding.

Step 1: do the sharding maths

Before sharding, estimate how many shards you need. The number comes from whichever limit you hit first: storage, write rate or read rate.

import math

def shards_needed(total_gb, writes_per_sec, reads_per_sec,
                  gb_per_shard=1_000, writes_per_shard=5_000, reads_per_shard=20_000,
                  headroom=0.5):
    # Plan to use only `headroom` of each limit, leaving room to grow
    by_storage = total_gb / (gb_per_shard * headroom)
    by_writes = writes_per_sec / (writes_per_shard * headroom)
    by_reads = reads_per_sec / (reads_per_shard * headroom)
    need = max(by_storage, by_writes, by_reads)
    limit = ["storage", "writes", "reads"][[by_storage, by_writes, by_reads].index(need)]
    return math.ceil(need), limit

# A messaging app: 3 years of messages
messages_per_day = 40_000_000
bytes_per_message = 500
total_gb = messages_per_day * 365 * 3 * bytes_per_message / 1e9
writes = messages_per_day / 86_400 * 3          # peak factor 3
reads = writes * 4

print(f"Data: {total_gb:,.0f} GB, peak writes {writes:,.0f}/s, peak reads {reads:,.0f}/s")
n, limit = shards_needed(total_gb, writes, reads)
print(f"Shards needed: {n} (limited by {limit})")

Data: 21,900 GB, peak writes 1,389/s, peak reads 5,556/s
Shards needed: 44 (limited by storage)

Here storage, not traffic, forces the split. Many teams round up to a power of two (64) or create many small logical shards up front, so that growth later means moving whole shards rather than re-splitting data. The per-shard limits in the code are made-up example numbers; in real life you measure them on your own hardware.

Three ways to partition

Range partitioning gives each shard a continuous range of keys: A-F, G-M, and so on, or January, February, March. Range queries are fast because neighbouring keys live together. The danger is a hot range: if the key is a timestamp, all of today's writes go to the newest shard.

Hash partitioning puts each key on the shard chosen by hash(key), usually with consistent hashing or fixed slots. Load spreads evenly, but range queries must visit every shard.

Directory (lookup) partitioning keeps a table that maps each key, or group of keys, to a shard. It is the most flexible: you can move one big customer to its own shard. The directory becomes a critical service that must be fast and highly available.

Step 2: see the time-based hot spot in action.

import hashlib
from collections import Counter

def hash_shard(key, n):
    return int(hashlib.md5(key.encode()).hexdigest(), 16) % n

def range_shard_by_day(day, days_per_shard=30):
    return day // days_per_shard

today = 365
writes_today = [f"order-{today}-{i}" for i in range(10_000)]

by_range = Counter(range_shard_by_day(today) for _ in writes_today)
by_hash = Counter(hash_shard(k, 8) for k in writes_today)

print("Range by date:", dict(by_range))
print("Hash, 8 shards:", dict(sorted(by_hash.items())))

Range by date: {12: 10000}
Hash, 8 shards: {0: 1311, 1: 1210, 2: 1278, 3: 1225, 4: 1290, 5: 1248, 6: 1248, 7: 1190}

All of today's 10,000 writes hit shard 12 under range partitioning, while hash partitioning spreads them almost evenly.

Choosing a good partition key

A good partition key has three qualities:

1. High cardinality: many distinct values, so data can spread. country is a poor key for an Indian app; almost everything is "IN".
2. Even load: no single value gets a big share of reads or writes.
3. Query locality: your most common queries include the key, so they touch one shard.

Hot partitions and celebrity keys

Even a good key can have a few huge values. A cricketer with millions of followers, a train on the Delhi-Mumbai route during Diwali, or one enterprise tenant with half of all data. That key's shard becomes a hot partition.

One fix is key salting: add a small suffix (0 to 9) to the hot key so its rows spread across several shards. Reads for that key must then query all the salted versions and combine them.

import hashlib
from collections import Counter

def shard(key, n=8):
    return int(hashlib.md5(key.encode()).hexdigest(), 16) % n

events = ["user:viratfan"] * 6000 + [f"user:{i}" for i in range(4000)]

plain = Counter(shard(k) for k in events)
salted = Counter()
for i, k in enumerate(events):
    if k == "user:viratfan":
        k = f"{k}#{i % 8}"                 # salt only the known hot key
    salted[shard(k)] += 1

print("Plain :", max(plain.values()), "writes on the busiest shard")
print("Salted:", max(salted.values()), "writes on the busiest shard")

Plain : 6475 writes on the busiest shard
Salted: 1991 writes on the busiest shard

Salting only the known hot keys keeps normal reads simple. Many systems keep a small list of "heavy" keys and treat them specially.

The costs of sharding

Sharding solves scale but adds real complexity. Know these costs before you shard:

• Cross-shard queries: a query without the partition key must ask every shard and merge results (scatter-gather). Its latency is set by the slowest shard.
• Cross-shard transactions: moving money between two users on different shards needs a distributed protocol or a different design, such as the saga pattern from Async Processing & Idempotency.
• Secondary indexes: an index on email is either local (each shard indexes its own rows, so lookups ask every shard) or global (a separate index, itself partitioned by email, which must be updated on every write).
• Unique IDs: auto-increment counters clash across shards. Use IDs that include a shard or machine number, or random IDs.
• Resharding: moving data while serving traffic is slow and risky. Plan many logical shards early so you can move them as whole units.

Common mistakes

• Sharding by a low-cardinality key such as country, status or gender.
• Sharding by time when writes are always "now", creating one hot shard.
• Forgetting the main query. If "my orders" does not include the partition key, every page view becomes a scatter-gather.
• Using hash % N with physical shards, making resharding move most of the data.
• Sharding too early, paying the complexity before it is needed.

Key points

• Sharding splits rows across databases by a partition key; replication copies the same rows.
• Estimate shards from storage, write and read limits, with headroom.
• Range partitioning suits range queries but risks hot ranges; hash partitioning spreads load; directories are flexible.
• A good partition key has high cardinality, even load and matches the main query.
• Hot keys can be salted; cross-shard queries, transactions and indexes are the main costs.

import math

def shards_needed(total_gb, writes_per_sec, reads_per_sec,
                  gb_per_shard=1_000, writes_per_shard=5_000, reads_per_shard=20_000,
                  headroom=0.5):
    by_storage = total_gb / (gb_per_shard * headroom)
    by_writes = writes_per_sec / (writes_per_shard * headroom)
    by_reads = reads_per_sec / (reads_per_shard * headroom)
    need = max(by_storage, by_writes, by_reads)
    limit = ["storage", "writes", "reads"][[by_storage, by_writes, by_reads].index(need)]
    return math.ceil(need), limit

total_gb = 2 * 10**7 * 2_000 / 1e9
print(total_gb, "GB")
print(shards_needed(total_gb, 300, 12_000))

40.0 GB
(2, 'reads')