Sharding Strategies

When to shard

One database is one checkout till.

Your system	Shard?	Why
200 GB, 2K writes a second	No	One tuned primary copes
2.5 TB, 3K writes a second	Not yet	Try a bigger machine, replicas, a cache
5 TB, 50K writes a second	Yes	One primary cannot take the writes

Choosing the shard key

The decides which queries stay on one and whether the load spreads, so it matters more than the scheme you hash it with.

Key	Verdict	Why
`user_id`	Good	Many values, spread evenly, one user's orders stay together
`country`	Weak	About 200 values, and one country can hold most users
`is_premium`	Bad	Two values means two shards, ever
`created_at`	Bad	Every new write lands on the newest shard

Share of writes on the busiest of 4 shards
	%
user_id, hashed	25%
tenant, one at 40%	40%
country, one at 60%	60%
created_at ranges	100%

Strategies and consistent hashing

A router (a library or a proxy) turns the into a shard, like a mail sorter reading the postcode.

Read more · 765 words

Play the four scenarios.

There are five ways to turn a shard key into a shard, and hash on user_id is the default answer. Pick another only when a specific need names it.

Strategy	How it works	Use it when	Costs
Hash mod N (the default)	`hash(key) mod N` picks the shard. Load spreads evenly because hashes look random, and a lookup by key touches one shard	You read and write by key and do not scan ranges: users, orders by user, sessions. MongoDB hashed shard keys work this way; Cassandra and DynamoDB hash partition keys too, though they use a ring or partitions rather than a plain mod	Range scans (every shard answers). Growth: plain mod moves N/(N+1) of keys when N changes
Range	Each shard holds a contiguous slice of the key space, so neighbours live together and a range scan stays on one shard	You scan by key interval and new keys do not arrive in order: `(sensor_id, time)`, alphabetical listings, a leaderboard cut into score bands. Bigtable, HBase and Spanner split ranges this way, automatically when a range grows	Hot last shard on timestamps or auto-increment ids, where every new write hits the last range
Directory	A table says which shard each key lives on, so you can place any key anywhere and move it later	A few keys are huge or special: a SaaS with 1 million small tenants and a handful 50 times the median, or a tenant that pays for isolation. 1 million tenants at about 50 bytes each is 50 MB, so the directory fits in memory	An extra lookup on every request (cache it, about 1 ms); a directory that must be replicated (a single point of failure); stale entries after a move. The shard must refuse a tenant that is no longer its own, so the app refreshes and retries
Geo	Each user's data lives in the region where they live, chosen once at signup. It is a directory whose key is a region	Law or latency demands it: data residency rules that keep EU data in the EU, or a round trip across the Atlantic (about 90 ms) against a local one (under 10 ms) felt on every click. Balance is not the goal, so a busy region simply gets more shards	Global queries that fan out across oceans, and users whose home region is wrong
Fixed slots (Redis Cluster)	Hash every key into a fixed number of slots (16,384 in Redis Cluster), then assign slots to machines. Adding a node moves whole slots, so keys are never rehashed	You want growth without consistent hashing: about the same 25% moved going from 3 to 4 nodes. Pick many more logical shards than machines, as the resharding section shows	Slots must be assigned and tracked

Play Add a shard to see the catch of plain mod.

. Picture a clock face. Nodes and keys both land on it, and a key belongs to the next node clockwise. Add a node and only its neighbours' keys move. Drag (extra points per node) down to 1 and watch the load drift uneven.

With hash % N, N/(N+1) of keys move when you add a node. A ring moves only 1/(N+1).

At 4 to 5 nodes, that is 80% versus 20%: the size of a 's miss storm. With 1 million keys, hash mod N moves 800,000 and the ring moves 200,000. Even doubling from 4 to 8 nodes moves half the keys with mod, because a key stays only when hash mod 4 equals hash mod 8.

gets the same 1/N result without a ring: every key scores each node with hash(node + key) and picks the highest. Removing a node moves only its own keys. It is simpler to code, and it costs one hash per node per lookup, fine for tens of nodes.

Tap any box to see what it is and what it talks to.

Service
API gateway
Database

Open in playground

Text description

The app saves an order for userId 42 through the shard router.
hash(42) mod 3 = 2, so the row goes to shard 2. Any query with the user id touches exactly one shard.
Shard 2 replicates to its own replica: every shard is a small primary/replica cluster.
Shard 2 confirms the commit.
Done. Later reads for user 42 run the same function and land on the same shard.

Tap any box to see what it is and what it talks to.

Service
API gateway
Database

Open in playground

Text description

Fetch user 42. Say hash(42) is 7,000,013.
7,000,013 mod 3 is 2, so the router asks shard 2 and nobody else.
One shard, one lookup. Even load is the strength of hashing.

Consistent hashing is a clock face of servers: each key walks clockwise to the next server it meets.

% of keys moved when going from N to N+1 nodes

hash % N
Consistent hashing

Show data table

% of keys moved when going from N to N+1 nodes
Nodes before adding one (N)	hash % N (%)	Consistent hashing (%)
2	66.7	33.3
3	75	25
4	80	20
5	83.3	16.7
6	85.7	14.3
7	87.5	12.5
8	88.9	11.1
9	90	10
10	90.9	9.1
11	91.7	8.3
12	92.3	7.7
13	92.9	7.1
14	93.3	6.7
15	93.8	6.3
16	94.1	5.9
17	94.4	5.6
18	94.7	5.3
19	95	5
20	95.2	4.8

Cross-shard queries, transactions and ids

Any query without the must ask every , and any write across two shards loses the database's .

Hot keys and resharding

A good can still have one scorching value, and then one machine melts while the others idle.

Code

Hash sharding

hash-shard.js

import { createHash } from 'node:crypto';
const hash = s => createHash('md5').update(s).digest().readUInt32BE(0);

// Hash the key and take it modulo the shard count. Even spread, but changing the count moves almost every key.
export const shardFor = (id, shardCount) => hash(String(id)) % shardCount;

// const db = pools[shardFor(userId, pools.length)];

Range sharding

range-shard.js

// Each shard owns a contiguous range of ids. Range scans stay on one shard,
// but sequential keys (auto-increment ids, timestamps) send every new write to the last shard.
const ranges = [{ max: 1e6, db: 'db0' }, { max: 2e6, db: 'db1' }, { max: Infinity, db: 'db2' }];
export const shardFor = id => ranges.find(r => id < r.max).db;

Directory (lookup) sharding

directory-shard.js

// A lookup table maps each tenant to a shard. The hot copy lives in Redis; the directory DB is the source of truth.
export async function shardForTenant(tenantId) {
  const shard = await redis.hget('shard-map', tenantId)
    ?? await directory.lookup(tenantId);
  return shard;
}
// Moving one huge tenant is a data copy plus one row update: directory.assign('acme', 'shard-7')

Consistent hashing

hash-ring.js

import { createHash } from 'node:crypto';
const hash = s => createHash('md5').update(s).digest().readUInt32BE(0);

export class HashRing {
  constructor(nodes = [], vnodes = 100) { this.vnodes = vnodes; this.ring = []; nodes.forEach(n => this.add(n)); }
  add(node) {
    for (let i = 0; i < this.vnodes; i++) this.ring.push({ pos: hash(`${node}#${i}`), node });
    this.ring.sort((a, b) => a.pos - b.pos);
  }
  remove(node) { this.ring = this.ring.filter(p => p.node !== node); }
  get(key) {                                        // binary search for the first point at or after hash(key)
    const h = hash(key);
    let lo = 0, hi = this.ring.length;
    while (lo < hi) {
      const mid = (lo + hi) >> 1;
      if (this.ring[mid].pos < h) lo = mid + 1; else hi = mid;
    }
    return this.ring[lo % this.ring.length].node;   // wrap around the ring
  }
}
// 4 → 5 nodes moves ~20% of keys (all of them to the new node); hash % N would move ~80%

Rendezvous hashing

rendezvous.js

import { createHash } from 'node:crypto';
const hash = s => createHash('md5').update(s).digest().readUInt32BE(0);

// Rendezvous (highest random weight) hashing: each key picks the node with the largest hash(node + key).
// No ring to maintain, and removing a node only moves that node's keys.
export const rendezvous = (key, nodes) => nodes.reduce((best, n) => (hash(n + key) > hash(best + key) ? n : best));

Online resharding

resharding.js

// Online resharding: never stop the world. The old shard stays the source of truth until reads flip.
export async function save(row) {
  await Promise.all([oldShard.write(row), newShard.write(row)]); // dual-write
}
// backfill in background, compare checksums, flip reads, delete old data

Scatter-gather

scatter-gather.js

// Scatter-gather: a query that can't use the shard key asks every shard, then merges, re-sorts and cuts
export async function topProducts(limit = 10) {
  const perShard = await Promise.all(shards.map(db =>
    db.query('SELECT id, name, sales FROM products ORDER BY sales DESC LIMIT $1', [limit])));
  return perShard.flatMap(r => r.rows).sort((a, b) => b.sales - a.sales).slice(0, limit);
}
// Latency = the slowest shard. If this query is frequent, maintain a precomputed global table instead.

Hot-key salting

hot-key-salting.js

// Hot-key salting: spread one celebrity counter over 16 sub-keys; reads add them up.
// (Separate GETs rather than MGET, because in Redis Cluster the sub-keys live in different slots.)
const SALTS = 16;
export const like = postId => redis.incr(`likes:${postId}:${Math.floor(Math.random() * SALTS)}`);
export async function likes(postId) {
  const parts = await Promise.all(Array.from({ length: SALTS }, (_, i) => redis.get(`likes:${postId}:${i}`)));
  return parts.reduce((sum, v) => sum + Number(v ?? 0), 0);
}

Snowflake ids

snowflake.js

// Snowflake-style 64-bit ids: unique across shards with no coordination, roughly time-ordered.
// 41 bits ms since a custom epoch (~69 years) | 10 bits worker | 12 bits sequence (4,096 ids/ms/worker)
const EPOCH = 1_704_067_200_000n;                       // 2024-01-01T00:00:00Z
export function snowflake(workerId) {
  if (!Number.isInteger(workerId) || workerId < 0 || workerId > 1023) throw new RangeError('workerId must be 0–1023');
  let lastMs = -1n, seq = 0n;
  return function next() {
    let now = BigInt(Date.now());
    if (now < lastMs) throw new Error('clock moved backwards');
    if (now === lastMs) {
      seq = (seq + 1n) & 4095n;
      if (seq === 0n) while ((now = BigInt(Date.now())) <= lastMs) {}   // sequence exhausted: wait for the next ms
    } else seq = 0n;
    lastMs = now;
    return ((now - EPOCH) << 22n) | (BigInt(workerId) << 12n) | seq;
  };
}
// Send ids to browsers as strings: 64-bit values don't fit in a JS number.

Sharding in the interview

Bring up only after you have shown that one cannot take the writes, then walk through five decisions in order.

If the interviewer says	You raise
"Writes outgrow the primary"	Shard by `user_id`, hash, after replicas and caching
"We need to scan by date"	Hash by id, sort by time inside the shard; or range with a hash prefix
"One customer is 40% of the data"	A directory and a dedicated shard
"Users must stay in their country"	Geo sharding with a home region
"How do you add a shard?"	Logical shards, then dual-write and backfill

Interview questions

What's the shard key for a multi-tenant SaaS?

tenant_id, with a directory so huge tenants get their own .

How do you run queries that span shards?

Keep them off hot paths: for admin, else a global index.

How do you reshard without downtime?

Add shards, dual-write, backfill, verify, flip reads, stop old writes.

Why do you need virtual nodes?

One point per node can leave a node with twice its share. Many points even it out and spread a dead node's keys over many survivors.

Hash or range sharding for time-series data?

Hash by series and sort by time inside each shard. Pure range-by-time sends every write to the newest shard.

Check yourself

Question 1 of 5

Score 2 out of 3 or better to mark this page complete.

1. Going from 4 to 5 nodes with hash % N moves about…20% of keys50% of keys80% of keys

Trade-offs

Bring it up before they ask

When to shard

Choosing the shard key

Strategies and consistent hashing

Consistent hashing ring

Cross-shard queries, transactions and ids

Hot keys and resharding

Add the new shards

Dual-write

Backfill

Verify

Flip reads

Stop old writes and clean up

Code

Hash sharding

Range sharding

Directory (lookup) sharding

Consistent hashing

Rendezvous hashing

Online resharding

Scatter-gather

Hot-key salting

Snowflake ids

Sharding in the interview

Prove one database is not enough

Choose the shard key from the main query

Pick the strategy

Handle the fallout

Say how you grow

Interview questions

Check yourself

Patterns behind this

Deep dives