Partitioning and sharding

Partitioning splits one dataset across many machines, so that each machine owns a subset of the keys. It is the only way to scale writes and storage past one machine. It is also the most expensive decision in a data design to change later. The partition key decides which queries stay cheap, which become fan-outs, and which shard catches fire when one key gets famous.

The problem it solves

An orders service runs on one PostgreSQL primary, with two replicas. The orders table holds 4 TB, and grows by 150 GB a month. Peak load is 9,000 writes a second. A well-tuned primary on large hardware holds roughly 10,000 to 20,000 small indexed writes a second. Past that, the WAL, checkpoints and vacuum fall behind. Three things now hurt together:

• Writes. Every write goes to the one primary. Replicas copy the writes. They cannot share them.
• Size. Rebuilding a replica or restoring from a 4 TB backup takes hours.
• Working set. The hot rows and indexes no longer fit in 512 GB of RAM, so p99 latency climbs.

Partitioning fixes all three: 8 shards of 500 GB, each primary taking about 1,100 writes a second. Each shard keeps its own replicas, so a node often leads some partitions and follows others.

The names vary by system: partition (Kafka, Cassandra, DynamoDB), shard (MongoDB, Elasticsearch), region (HBase), tablet (Bigtable), range (CockroachDB). Here “shard” means a machine or replica group and “partition” the unit of data that moves between them.

Shard last

Sharding takes away cross-shard joins, cross-shard transactions, global unique constraints and simple operations, so each cheaper step comes first:

Before sharding, in this order

Scroll or pinch to zoom, drag to pan

• Color key
• Database
• Service
• Cache

Splitting by function (vertical partitioning) puts different tables on different databases. It often buys a year, but a table that outgrows a machine still needs horizontal partitioning: the rows of one table split by key. That is the rest of this page.

The triggers are measurable:

• Sustained writes above about 60% of one primary’s capacity.
• A replica rebuild or restore that takes longer than the recovery time objective. For PostgreSQL or MySQL, that’s often around 1 to 2 TB.
• A hot working set that no longer fits in RAM.

Read load alone is never a reason to shard.

Common mistake

"We shard to handle more reads"

Read load is answered by replicas and caches, which are cheap and keep every query working. Sharding splits write load and storage. A read-heavy service that shards turns a working JOIN into a fan-out and gains nothing a replica could not give. Shard when writes or size, measured, have outgrown one primary.

How it works

Every scheme answers one question: given a key, which partition owns it? Each step below is forced by a flaw in the one before.

Step 1: Ranges of the key

Sort the keys, and cut the sorted order into contiguous ranges. a to f goes on shard 1, g to m on shard 2, and so on. A small table of boundaries says who owns what. This is how Bigtable, HBase, Spanner, CockroachDB and MongoDB’s ranged sharding work.

The strength is order. With a key of (sensor, time), “sensor 17 from 09:00 to 10:00” reads one contiguous slice. Ranges also split naturally. Past a threshold, the store cuts one in two, and can move a half. The default threshold is 10 GB per HBase region, and 512 MiB per CockroachDB range.

The flaw is skew. If keys arrive in order (timestamps, auto-increment IDs, anything prefixed by the date), every new write lands on the last range. One shard takes all the writes, and splitting does not help: the new last range is just as hot.

Step 2: Hash the key

Hash the key and use the hash to pick the shard. A good hash scatters neighbouring keys, so order:1000 and order:1001 land on unrelated shards, and sequential inserts spread evenly. This is the default in DynamoDB, Cassandra, Redis Cluster and Kafka.

Hashing gives up order: “orders from the last hour” must ask every shard. Many stores keep both, with a compound key. A hash-partitioned partition key picks the shard. A sort key (clustering key) orders rows inside it. Cassandra’s PRIMARY KEY ((channel_id, bucket), message_id), and DynamoDB’s partition key plus sort key, have this shape. Range queries then work inside one partition key, which is often all that is needed.

The simplest way to turn a hash into a shard is hash(key) mod N. That is the next flaw.

Step 3: Fixed slots and a map

With hash(key) mod N, changing N changes the answer for almost every key. Going from 4 to 5 shards moves about 80% of keys (N / (N + 1) in general). Only 20% needed to move, to fill the new shard.

Kafka shows the symptom in production. Its default partitioner is a hash modulo the partition count. So adding partitions to a topic sends existing keys to new partitions. That breaks per-key ordering across the change.

The fix is a level of indirection. Hash keys into a fixed number of slots, chosen once and much larger than the shard count: 4,096, or 16,384 as in Redis Cluster. A small map assigns slots to shards. The hash never changes, so a key never changes slot. Adding a shard means reassigning some slots in the map and copying only those slots’ data.

Two steps from key to shard: a fixed hash, then an editable map

Scroll or pinch to zoom, drag to pan

• Color key
• Service
• Database

The slot count is the ceiling on shards, so choose it generously: at 8 shards, 4,096 slots give 512 each. Postgres schemas, MySQL databases or Vitess key ranges can play the role of slots.

The code below routes through a slot map and compares the keys moved by adding a shard with the keys moved by mod N. It prints 20.2% against 80.0% for 100,000 keys.

JSlotRouter.javaPyslot_router.pyC#SlotRouter.csGoslot_router.goTSslot-router.ts

JavaPythonC#GoTypeScript

JavaPythonC#GoTypeScript

import java.nio.charset.StandardCharsets;
import java.util.ArrayList;
import java.util.List;

// Hash slots with an explicit slot map: adding a shard moves only the slots it takes.
class SlotRouter {
    static final int SLOTS = 4096; // fixed forever: keys hash to slots, never directly to shards

    private final int[] owner = new int[SLOTS];
    private int shards;

    SlotRouter(int shards) {
        // Deal the slots round robin, so every shard starts with an equal share.
        for (int s = 0; s < SLOTS; s++) owner[s] = s % shards;
        this.shards = shards;
    }

    // A stable hash. Never use String.hashCode() across languages or versions you do not
    // control: every client must send the same key to the same shard.
    static long fnv1a(String key) {
        long h = 0x811C9DC5L;
        for (byte b : key.getBytes(StandardCharsets.UTF_8)) {
            h ^= (b & 0xFF);
            h = (h * 0x01000193L) & 0xFFFFFFFFL;
        }
        return h;
    }

    // Two steps. The hash picks a slot and never changes; the map picks the shard and is the
    // only thing a resize edits. Clients cache the map and refresh it on a redirect.
    int route(String key) {
        return owner[(int) (fnv1a(key) % SLOTS)];
    }

    void addShard() {
        int added = shards++;
        List<List<Integer>> held = new ArrayList<>();
        for (int s = 0; s < added; s++) held.add(new ArrayList<>());
        for (int slot = 0; slot < SLOTS; slot++) held.get(owner[slot]).add(slot);
        // The new shard takes its fair share, one slot at a time from whichever shard holds the
        // most. Only those slots move, and the layout stays even.
        for (int k = 0; k < SLOTS / shards; k++) {
            int donor = 0; // ties go to the lowest id
            for (int s = 1; s < added; s++) if (held.get(s).size() > held.get(donor).size()) donor = s;
            List<Integer> mine = held.get(donor);
            owner[mine.remove(mine.size() - 1)] = added;
        }
    }
}

class Main {
    public static void main(String[] args) {
        int n = 100_000;
        String[] keys = new String[n];
        for (int i = 0; i < n; i++) keys[i] = "user:" + i;

        SlotRouter router = new SlotRouter(4);
        int[] before = new int[n];
        for (int i = 0; i < n; i++) before[i] = router.route(keys[i]);
        router.addShard();
        int moved = 0;
        for (int i = 0; i < n; i++) if (router.route(keys[i]) != before[i]) moved++;
        System.out.printf("slot map, 4 -> 5 shards: %.1f%% of keys moved%n", moved * 100.0 / n);

        // The naive layout for comparison: the shard count is inside the formula, so changing it
        // changes the answer for most keys.
        int naive = 0;
        for (String k : keys) if (SlotRouter.fnv1a(k) % 4 != SlotRouter.fnv1a(k) % 5) naive++;
        System.out.printf("hash mod N, 4 -> 5 shards: %.1f%% of keys moved%n", naive * 100.0 / n);
    }
}

Try it

1,000 keys take writes under three workloads: popular users scattered through the key space, one celebrity key with 30% of all writes, and time-ordered inserts. Pick a scheme, and add or remove shards. Bars show each shard’s share of writes, against its fair share. The grid shows which shard owns each key. After a change, it shows which keys moved.

Workload

Popular usersOne celebrity keyTime-ordered inserts

Partition by

Key rangehash mod NHash slotsSlots + salting

Shards: 4

Add a shardRemove the newestReset

Hottest shard

1.0×

of its fair share

Last change moved

none

add or remove a shard

1,000 keys in key order, colored by owning shard. After a change, keys that did not move fade.

Popular users on key ranges. Add a shard and watch which keys move.

Things to find:

• Time-ordered inserts on key ranges put every write on the last range. And a split only creates a new last range.
• hash mod N is even, but moves about 80% of keys per added shard. Slots move about 1/N.
• No scheme that places whole keys spreads the celebrity. Salting spreads its writes, but then every read of it asks 8 sub-keys.

Step 4: A directory

Hashing decides placement by formula. A directory decides it by lookup. It’s a table that maps each key, or each tenant, to a shard. Any placement is possible. A big tenant can get a dedicated shard. A tenant can move between shards, by copying its data and updating one row. And placement can follow geography, for data residency.

The cost is a lookup on every request, and a component that must never be wrong or down. So the directory is cached in every router, with a version number. A slot map is a directory at slot granularity. A per-key directory earns its cost when a few keys need their own placement. Examples are SaaS tenants of very different sizes, or stock symbols placed by measured message rate.

Step 5: Routing

Something has to turn a key into a network address. The choice is about hops and who must know the map.

Three places to route a request

Scroll or pinch to zoom, drag to pan

• Color key
• Service
• Database

Routing	Extra hops	Who holds the map	Wins when	Examples
Client library	0	Every client	Latency matters and you control every client	Redis Cluster clients, Cassandra drivers
Proxy tier	1, about 0.2 to 0.5 ms in a DC	The proxies	Many languages, or clients you cannot upgrade	MongoDB mongos, Vitess VTGate, Twemproxy
Any node coordinates	0 or 1	Every node, learned by gossip	Peer-to-peer stores with no special tier	Cassandra, ScyllaDB, Riak

Every router’s map is a cached copy of an authority, such as etcd, ZooKeeper, MongoDB’s config servers, or gossip. Copies go stale during a move. So a shard asked for a slot it no longer owns must reply with a redirect (Redis Cluster’s MOVED). It must never answer from data it does not own.

Where consistent hashing fits

Consistent hashing is the other answer to mod N. Shards and keys hash onto a ring. A key belongs to the next shard clockwise. A new shard takes about 1/N of keys, with no central map. Dynamo, Cassandra and client-side cache rings use it. A slot map moves the same 1/N, but makes every key’s location explicit. That’s why most databases that migrate data use slots or ranges. The consistent hashing page covers rings, virtual nodes, jump and rendezvous hashing.

Options and tradeoffs

Range, hash or directory

Scheme	Range scans	Even writes	Adding capacity	Wins when
Range	Yes, one slice	Only if keys arrive in random order	Split a range, move half	Queries scan by key order: time series per entity, prefix search
Hash with slots	Only inside a compound key	Yes, for any key order	Move slots, about 1/N of data	Point lookups by ID: the default for most designs
Hash mod N	No	Yes	Moves about N/(N+1) of data	The shard count will truly never change, such as a fixed cluster
Directory (per key)	Depends on placement	Only as good as the placement policy	Move one tenant, update one row	A few tenants dwarf the rest, or placement must follow residency

TradeoffHash with slots as the default

We gain

Even load whatever the key order, and single-shard point lookups. Growth moves only the slots you choose. And every key’s location is visible in one small map.

We give up

No range scans across keys. A map that every router must cache and refresh. And a slot count that caps the shard count, and is painful to change later.

Choosing the shard key by access pattern

The shard key is chosen from the queries, not from the data. List the queries with their rates, then pick the key that makes the hottest ones single-shard and spreads writes evenly. Every query without the shard key becomes a fan-out, a second copy of the data, or both.

Take an e-commerce orders table with three hot queries:

• Fetch an order by ID (40,000 a second).
• List a customer’s recent orders (8,000 a second).
• The seller dashboard’s orders for one seller today (500 a second).

Shard key	Order by ID	Customer’s orders	Seller’s orders today	Problem
Order ID (hash)	One shard	Every shard	Every shard	The second-hottest query is a scatter-gather
Created time	One shard	Every shard	One shard	All writes go to the newest range
Seller ID	Every shard, unless encoded	Every shard	One shard	One big seller is one hot shard
Customer ID	One shard, if ID encodes it	One shard	Every shard (rare)	Seller view needs a second copy keyed by seller

The pick here is customer ID, with the customer’s shard embedded in each order ID. Then a lookup by ID needs no directory. Instagram did exactly this in 2012. Each 64 bit ID holds 41 bits of time, 13 bits of logical shard, and 10 bits of sequence. So any media ID names its shard. The seller dashboard reads a second table keyed by seller. It’s fed asynchronously from the orders’ change stream.

Three rules carry most decisions:

• High cardinality. The key needs far more distinct values than shards. Country code has about 200 values and the largest country is a hot shard by itself.
• Even access, not just even count. A million users with one celebrity is uneven. Check the top 1% of keys by traffic, not the number of keys.
• Bound the partition’s size. Anything that grows forever under one key gets a time or sequence bucket in the key. Discord keys messages by (channel_id, bucket) with fixed time buckets, and Cassandra guidance keeps partitions well under 100 MB.

TradeoffShard by the entity the hot queries name

We gain

The hottest reads and every write for one entity stay on one shard, with local transactions and local joins inside it.

We give up

Queries keyed by anything else need a fan-out, or a second copy kept in sync. And an entity can never change shards cheaply, once its shard is in its IDs.

Secondary indexes: local or global

A secondary index answers a query by a field that is not the shard key, such as “orders with status FAILED”. Partitioned stores offer two shapes.

A local index lives on each shard, and covers only that shard’s rows. The row and its index entry commit together, so writes stay cheap. But a read by the indexed field must ask every shard and merge the answers. This is called scatter-gather. Cassandra secondary indexes, and MongoDB indexes on non-shard-key fields, work this way.

A global index is itself partitioned by the indexed value. So all FAILED entries live together, and a read asks one index shard. But a write now touches two shards. So the index either lags, or costs a distributed commit. DynamoDB global secondary indexes are asynchronous. Spanner and CockroachDB keep them transactional.

A query by a non-key field with a local index

Scroll or pinch to zoom, drag to pan

• Color key
• Database
• Service

The response waits for the slowest shard, so tail latency compounds. Say each shard is slower than its p99 one time in a hundred. Then a request to 100 shards meets a slow one 63% of the time (1 minus 0.99 to the power 100). Every shard also works on every fan-out query. So adding shards adds capacity for keyed queries only.

Index shape	Write cost	Read by indexed field	Consistency	Wins when
Local	One shard, same transaction	Every shard, wait for slowest	Same as the row	The indexed query is rare, or always also has the key
Global, asynchronous	Row now, index a moment later	One index shard, then the rows	Lags by ms to seconds	Frequent reads by the field that tolerate brief lag
Global, transactional	A cross-shard commit per write	One index shard, then the rows	Immediate	Uniqueness checks or reads that must see the last write
Separate copy keyed by field	An async pipeline you run	One shard of the copy	Lags	The second access pattern is hot: a seller view, likes

TradeoffLocal indexes plus a copy for the one hot secondary query

We gain

Writes stay single-shard and cheap, and rare queries tolerate a fan-out. The one hot query by another field gets its own partitioned copy, with single-shard reads.

We give up

A pipeline to run and monitor. Lag between the copies. And a fan-out cost that grows with the shard count, for every query left on local indexes.

Cross-shard queries and transactions

Inside one shard, transactions, joins, unique constraints and foreign keys all still work. Across shards, each needs a plan.

Need across shards	Options	What it costs
A join	Co-locate both tables by the same key, or denormalize, or join in the app	Co-location fixes one key; app joins are N queries
An aggregate (count, top-k)	Scatter-gather with partial aggregates, or a precomputed rollup	Fan-out per query, or a pipeline and lag
A unique constraint (email)	A separate table partitioned by the unique value, written first	Two writes and a cleanup path when the second fails
An atomic write on two shards	Two-phase commit, or a saga with compensations	2PC blocks on a coordinator failure; sagas expose middle state

The first answer is the shard key. A payment and its ledger lines share a payment ID. A booking and its inventory row share a hotel ID. Sometimes two shards must change together, such as a transfer between accounts. Then the options are two-phase commit or a saga. Spanner and CockroachDB do distributed transactions natively. They pay extra consensus round trips per commit: milliseconds across zones, and 100 ms or more across continents.

TradeoffKeep transactions inside one shard by key choice

We gain

Local ACID transactions at single-database latency, no coordinator to fail, and no in-between states for users to see.

We give up

The shard key is bent toward the transaction boundary. And the rare operation that spans entities needs a saga or an outbox, which is real code to write and test.

Hot keys: salt, split, cache or isolate

Hashing spreads keys, not traffic. A key with 30% of the traffic puts 30% on one shard, under any scheme. A DynamoDB partition serves at most 3,000 reads and 1,000 writes a second. One Redis shard serves about 100,000 to 200,000 simple operations a second. So a celebrity key hits a ceiling long before the cluster is busy.

Fix	How it works	Read cost	Wins when
Salt (split the key)	Write key#0 to key#7, chosen by a hash of the writer	Read all 8 sub-keys and merge	Write-hot counters and append streams
Replicate the key	Store full copies under several keys, read any one	One read; writes go to all copies	Read-hot, rarely written values
Cache in front	An in-process cache with a short TTL absorbs repeat reads	Stale by up to the TTL	Read-hot values that tolerate seconds of staleness
Split or isolate	Split a hot range, or give a hot tenant its own shard by directory	None	The hot unit is a range or a tenant, not one key

Salting must be deterministic. A salt from the writer’s ID sends a retried write to the same sub-key, so deduplication and replay still work. A random salt breaks both. Apply it only to keys detected as hot, because every read of a salted key becomes a fan-out of 8.

JShardedStore.javaPysalted_counter.pyC#ShardedStore.csGosalted_counter.goTSsalted-counter.ts

JavaPythonC#GoTypeScript

JavaPythonC#GoTypeScript

import java.nio.charset.StandardCharsets;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

// Stands in for a sharded database: one map per shard, and a write count per shard.
class ShardedStore {
    final List<Map<String, Integer>> data = new ArrayList<>();
    final int[] writes;

    ShardedStore(int shards) {
        for (int s = 0; s < shards; s++) data.add(new HashMap<>());
        writes = new int[shards];
    }

    // A stable hash, the same on every app server.
    static long fnv1a(String key) {
        long h = 0x811C9DC5L;
        for (byte b : key.getBytes(StandardCharsets.UTF_8)) {
            h ^= (b & 0xFF);
            h = (h * 0x01000193L) & 0xFFFFFFFFL;
        }
        return h;
    }

    int shardOf(String key) {
        return (int) (fnv1a(key) % data.size());
    }

    void incr(String key) {
        int s = shardOf(key);
        data.get(s).merge(key, 1, Integer::sum);
        writes[s]++;
    }

    int get(String key) {
        return data.get(shardOf(key)).getOrDefault(key, 0);
    }
}

// Salting a hot key: one logical counter written as several sub-keys on different shards.
class SaltedCounter {
    private final ShardedStore store;
    private final int ways; // 1 means no salting; raise it only for keys detected as hot

    SaltedCounter(ShardedStore store, int ways) {
        this.store = store;
        this.ways = ways;
    }

    private String subKey(String key, long i) {
        return ways == 1 ? key : key + "#" + i;
    }

    // The salt comes from the writer, not from a random number: a retried write lands on the
    // same sub-key, so per-sub-key dedup and replay still work.
    void incr(String key, String writer) {
        store.incr(subKey(key, ShardedStore.fnv1a(writer) % ways));
    }

    // The price of salting: every read asks all the sub-keys and adds them up.
    int read(String key) {
        int total = 0;
        for (int i = 0; i < ways; i++) total += store.get(subKey(key, i));
        return total;
    }
}

class Main {
    public static void main(String[] args) {
        for (int ways : new int[] {1, 8}) {
            ShardedStore store = new ShardedStore(8);
            SaltedCounter likes = new SaltedCounter(store, ways);
            for (int u = 0; u < 80_000; u++) likes.incr("post:42:likes", "user:" + u);
            int busiest = Arrays.stream(store.writes).max().getAsInt();
            System.out.printf(
                    "ways=%d: total=%d, busiest shard took %d writes, a read touches %d key(s), writes per shard %s%n",
                    ways, likes.read("post:42:likes"), busiest, ways, Arrays.toString(store.writes));
        }
    }
}

Run it. 80,000 likes on one post put 80,000 writes on one of 8 shards. Salted 8 ways, each shard takes 10,000, and a read sums 8 sub-keys.

TradeoffSalt only the keys that are hot

We gain

A single key’s writes spread over many shards, so one celebrity no longer caps throughput at one partition’s limit.

We give up

Reads of that key fan out and merge, and ordering across sub-keys is lost. And a detector must decide which keys are hot, and tell every writer and reader.

Resharding without downtime

Moving data between shards while traffic flows is where partitioning goes wrong in practice.

Approach	How capacity grows	Data moved per change	Used by
Many fixed partitions from day one	Move whole partitions or slots to new nodes; the key-to-partition map never changes	The moved partitions only	Redis Cluster, Elasticsearch, Riak, Vitess, Couchbase
Dynamic range splitting	The store splits a range that grows too big or too hot and moves half	Half of one range	Bigtable, HBase, Spanner, CockroachDB, MongoDB
Migrate to a new layout	Dual writes, backfill, verify, cut reads, cut writes	Everything, once	Moving from one database to N, or changing the key

The first is the cheapest to run, with one limit. The partition count is fixed up front, the way Elasticsearch’s primary shard count is fixed at index creation. Too few means a split or a reindex later. The third is how a single database becomes a sharded one. It goes in phases, and each phase can be rolled back:

Dual writes, backfill and cutover for one slot

Scroll or pinch to zoom, drag to pan

• Color key
• Background worker
• Database

Three details decide whether it works:

• The backfill never overwrites a newer dual-written row. It inserts if absent, or compares versions.
• Rows that are updated, not just inserted, need a version per row. Or use the source’s change stream (CDC) instead of dual writes, so the last writer wins.
• The write cutover needs a short per-slot fence. Pause that slot’s writes for a few hundred milliseconds, drain, flip the map, and resume. Then no write lands on the old copy after the flip.

The URL shortener’s online repartitioning walkthrough steps through all six phases for a 3 TB table, with the checks that make each one safe.

TradeoffMany small partitions, chosen up front

We gain

Growth means moving partitions from busy nodes to new ones, with no change to where any key hashes. So a resize is routine operations, not a migration project.

We give up

Per-partition overhead (files, connections, metadata) while the cluster is small, and a ceiling on nodes equal to the partition count.

Go deeperResharding in specific systems2 min

System	Unit that moves	How a move works
Redis Cluster	One of 16,384 hash slots	Slot marked migrating on the source and importing on the target; keys move one by one; clients follow ASK then MOVED redirects
Vitess	A key range of keyspace IDs	VReplication copies and streams changes to target shards; SwitchTraffic flips reads, then writes; reversible
MongoDB	A chunk of the shard key range (128 MB default)	The balancer migrates chunks between shards; documents copied, then changes caught up, then ownership committed on the config servers
CockroachDB	A range, 512 MiB by default	Ranges split by size or load; replicas rebalance through Raft by adding a replica and removing another
DynamoDB	A partition	Splits automatically by size (about 10 GB) or sustained heat; no operator action, but a single item cannot split
Kafka	A topic partition	Reassignment copies replicas to new brokers; adding partitions changes key placement, so keyed topics are sized up front

How it fails

Failure	What you see	Blast radius	Fix
Hot key	One shard’s CPU or throttling high while the rest idle; p99 up for keys on it	Every key that shares the shard	Salt, replicate or cache the key; detect top keys continuously
Monotonic key on ranges	All inserts on the last range; splits do not help	All writes	Hash or prefix the key; for time series, partition by entity then time
Unbounded partition	One partition grows to many GB; compaction and repair stall on it	That partition, then its node	Add a bucket (time or sequence) to the partition key
Scatter-gather tail	Fan-out queries’ p99 tracks the slowest shard and rises with shard count	Every fan-out query, and shard CPU for all	Make the hot query keyed; global index or copy; hedged requests
Stale routing map	Requests to a shard that no longer owns the slot	Requests for moved slots during a resize	Versioned map; shards reject with a redirect, clients refresh
Lost or stale writes in a move	Checksums differ after backfill; old values reappear after cutover	Rows written during the move	Insert-if-absent or version compare; per-slot write fence at cutover

Go deeperThree more failure modes1 min

Failure	What you see	Blast radius	Fix
Partial cross-shard write	Money left one account and never arrived; orphaned index entries	The entities in the failed operation	Saga with compensations, outbox, or 2PC; reconcile daily
Global index lag	A just-written item missing from a query by another field	Readers of the index for the lag window, ms to s	Read the base table for read-your-writes; alert on lag
Rebalancing storm	A node fails, data copies saturate the network, healthy nodes slow down	The whole cluster	Throttle moves; wait minutes before moving data for a node that may return

A hot key is the most common of these, and it hides in averages. A 20-shard cluster at 30% average CPU can have one shard at 100%. Watch per-shard load against the mean.

Common mistake

"Hashing the key fixes hot spots"

Hashing evens out how many keys each shard holds. It does nothing for one key that gets a large share of requests, because a hash sends every request for that key to the same place. Hot keys need salting, replication or caching of that key. In the demo, only salting spreads the celebrity’s writes.

At scale

Quantity	Typical figure
Comfortable size of one PostgreSQL or MySQL shard	About 1 to 2 TB, limited by backup, restore and replica rebuild time
Writes one well-tuned relational primary holds	About 10,000 to 20,000 small indexed writes a second
DynamoDB partition limits	3,000 read units and 1,000 write units a second, about 10 GB
CockroachDB range, HBase region, MongoDB chunk	512 MiB, 10 GB, 128 MB by default
Redis Cluster	16,384 slots; one shard about 100,000 to 200,000 simple operations a second
Cassandra partition	Keep well under 100 MB; virtual nodes per node default to 16 since 4.0
Proxy hop (mongos, VTGate)	About 0.2 to 0.5 ms inside one data center
Fan-out to 100 shards, each 1% slow	63% of requests wait for at least one slow shard

At 10 times, more shards make hot keys relatively worse. A key with 5% of traffic is under half a shard’s fair share on 8 shards, and four times it on 80. Fan-out queries cost 10 times the work. So the ones that stay hot get keyed copies. And resizes become routine enough to need automated, throttled, reversible tooling.

At 100 times, placement becomes a service. Tenants are placed by measured load. Hot-key detection feeds salting and caching. Moves happen every day. And cells (groups of shards with their own routing) mean one bad deploy or runaway tenant reaches one cell, not the fleet.

What to monitor:

• Per-shard request rate, CPU and p99, against the fleet mean. Alert at 1.5 times the mean.
• Per-shard size and growth, forecast against the split or move threshold.
• Top keys by request rate, sampled at the router or with a count-min sketch.
• Fan-out queries’ share and p99, apart from keyed queries.
• Redirect rate (MOVED replies) and map version per router, as a staleness signal.
• During a move: backfill progress, replication lag, checksum and shadow-read mismatches.

In the interview

Interviewer asksYour users table is sharded by user ID. Product wants login by email. What do you do?

Login is hot, so I would not scatter-gather it across every shard. I would add a small table, partitioned by a hash of the email, that maps email to user ID. That’s two single-shard reads. It also enforces email uniqueness, which a per-shard unique index cannot. Signup writes the email row first, with an insert-if-absent, then the user row. If the second write fails, a cleanup job removes orphaned email rows older than a few minutes. Changing an email is a move between two email shards. So it is a small saga: claim the new email, update the user, release the old one.

Interviewer asksOne tenant is 40% of all traffic and growing. Hash partitioning by tenant ID is melting one shard. Fix it.

The tenant is the hot unit, so I change the key for that tenant only. If its data divides, I partition it by tenant plus project, or tenant plus a hash of the record ID. I record that in a directory entry. Its rows spread over many slots, and only its tenant-wide queries become fan-outs. If the data does not divide, I give the tenant dedicated shards through the directory, and scale them up. Either way, the move uses dual writes and a backfill for that tenant only. And I add a per-tenant quota, so the next big tenant is noticed before it melts a shard.

Interviewer asksWhy not just use 1,024 shards from the start so you never reshard?

Physical shards have a fixed cost each: a primary and replicas, backups, connections, monitoring. 1,024 small databases for a 2 TB dataset is a large bill and an operations burden. What I want from day one is 1,024 or 4,096 logical partitions, mapped onto however many physical shards the load needs today, perhaps 4. Growth is then moving logical partitions onto new machines, which needs no rehashing and no change to any key. I would also keep the shard count out of the IDs and the hash function, so only the map changes.

Common mistake

Using an auto-increment ID or a timestamp as a range shard key

Sequential keys look evenly distributed when you count them, and they are, across all time. At any moment, though, every insert carries the newest value and lands on the last range, so one shard takes 100% of writes. Hash the key, prefix it with something high-cardinality such as the entity ID, or use a hash-partitioned store. Range keys on time work only when time is the second part of a compound key, as in (sensor_id, time).

Mid-level covers

Knows when sharding is needed, and that read replicas come first. Picks hash or range with a reason. Chooses a key that keeps the hot queries on one shard. Names hot keys and cross-shard queries as the costs.

Senior adds

Lists the queries with rates before choosing the key. Separates logical partitions from physical shards. Prices scatter-gather in tail latency. Handles the queries the key does not serve, with indexes or copies. Salts only detected hot keys. Plans the move as reversible phases, with checks and a write fence.

Seen in

• URL shortener: codes hashed into 4,096 slots, an owner table for the one unkeyed query, and the online move to 16 shards.
• Instagram: metadata sharded by user, with 13 bits of logical shard in every ID so a lookup needs no directory.
• Chat: messages keyed by conversation plus a sequence bucket, which bounds each partition at about 1.5 MB.
• Key-value store: why mod N fails, and a ring with virtual nodes and gossip routing.
• Distributed cache: a client-side ring with 160 virtual nodes per shard, and hot keys replicated.
• Rate limiter: counters sharded by API key with a hash tag, so one request’s buckets sit on one shard.
• Message queue: partitions keyed by the entity whose order matters, and a finer key, not random salt, for a hot tenant.
• Job scheduler: runs partitioned by a hash of the job ID, never by due time, so a run’s whole life is one shard.
• Hotel reservation: sharded by hotel through 4,096 slots so every booking transaction is local, with a guest table fed from the outbox.
• Stock exchange: order books partitioned by symbol through a directory placed by measured message rate.
• Leaderboard: scores sharded by a hash of the player ID, with the top page merged from every shard’s full top k.
• Top-k heavy hitters: views partitioned by item, which makes each item’s count exact on one partition.
• Ad click aggregator: counts keyed by ad, with a deterministic salt from the impression ID for hot ads.
• Web crawler: the frontier partitioned by site into 4,096 slots, so politeness stays local to one node.
• Proximity service: the counter-example, a 6 GB geo index copied to every server instead of sharded.

Recap

When. Shard when measured writes or size outgrow one primary. Fix queries, add replicas and caches, scale up and split by function first.

How. Range keeps order and suffers sequential keys. Hash spreads any key order and loses range scans. Hash into fixed slots mapped to shards, so growth moves about 1/N of the data. A directory places tenants explicitly at the cost of a lookup.

The key. Chosen from the queries and their rates: hottest reads and every transaction single-shard, high cardinality, even traffic, bounded partitions.

Costs. Local indexes mean scatter-gather, whose latency is the slowest shard’s. Global indexes mean two-shard writes and lag. Cross-shard atomicity needs 2PC or a saga.

Hot keys. Hashing does not spread one key. Salt write-hot keys deterministically, replicate or cache read-hot ones, isolate hot tenants.

Moves. Many logical partitions up front. Dual writes, backfill with insert-if-absent, verify, flip reads, fence and flip writes, each phase reversible.

Further reading

• Consistent hashing, replication, distributed transactions and sagas and unique ID generation on this site.
• Designing Data-Intensive Applications, Martin Kleppmann, chapter 6: the clearest treatment of partitioning schemes, secondary indexes and rebalancing.
• Sharding & IDs at Instagram, Instagram engineering: logical shards on Postgres schemas and IDs that carry their shard.
• How Discord stores trillions of messages, Discord engineering: bucketed partition keys, hot partitions and a live migration.