Sharding & partitioning · How it works

9 min read
Senior15 min read
Rapid overview

How it works

Why shard, and why not yet

Sharding is how you scale writes and storage beyond one node. Reads scale earlier and more cheaply with replicas and caches. The costs are permanent: cross-shard joins and transactions become hard, unique constraints need a global mechanism, operational work multiplies per shard, and a bad shard key is expensive to change.

Q: What should you try before sharding a relational database?

A: Everything that scales without changing the data model: tune queries and indexes, add read replicas and a cache for read load, move large blobs to object storage, archive or partition cold data by time within the same database, scale the primary vertically (modern single nodes handle terabytes and tens of thousands of writes per second), and split by function — separate databases per service or bounded context. Shard when write throughput or dataset size genuinely exceeds a single primary with headroom, because sharding makes joins, transactions, uniqueness and operations permanently harder.

Three ways to map keys to shards

flowchart TD K[Record key] --> H{Partitioning strategy} H -->|hash| HS[hash of key mod N or ring position] H -->|range| RS[key falls in a sorted range - A to F, G to M] H -->|directory| DS[lookup table maps key or tenant to shard] HS --> HP[Even spread, no range scans] RS --> RP[Range scans, risk of hot tail] DS --> DP[Any placement, extra lookup and a critical service]
  • Hash: shard = hash(key) mod N, or a position on a hash ring. Spreads load evenly and destroys locality: a query for a range of keys hits every shard.
  • Range: shards own contiguous key ranges (HBase, Bigtable, CockroachDB, MongoDB ranged). Efficient range scans; sequential keys such as timestamps or auto-increment ids send all new writes to the last range — a hot tail. Ranges split automatically as they grow.
  • Directory (lookup): a mapping service records which shard owns each key or tenant. Any placement is possible, big tenants can be moved individually, at the cost of a lookup (cached) and a highly available directory.
Q: Why does range partitioning on a timestamp create a hot spot, and how is it avoided?

A: New writes always carry the newest timestamp, so they all fall into the last range and one shard takes the entire write load while the others sit idle on historical data. Avoid it by prefixing the key with something that spreads writes — a hash of the device or user id, or a small bucket number — so writes for the same moment land on many shards, and use a compound key such as (sensor id, timestamp) so per-sensor time ranges are still contiguous. The trade-off is that a query for "all events in the last minute" must now read every shard.

Why hash mod N breaks when N changes

With hash(key) mod N, adding a fifth shard to four changes the result for about 80% of keys, so almost all data must move. Two fixes:

  1. Consistent hashing: shards and keys are placed on a ring; a key belongs to the next shard clockwise. Adding a shard takes over only the arc before it, about 1/N of the keys.
  2. Fixed virtual partitions: create many more partitions than nodes up front (for example 1,024) and assign partitions to nodes. Adding a node moves whole partitions; keys never rehash. Used by Cassandra's vnodes, Elasticsearch, Riak, Couchbase and Redis Cluster (16,384 slots).
Q: Why does hash mod N cause massive data movement when a shard is added?

A: Because the modulus changes the shard of almost every key: a key maps to a new shard unless hash mod N and hash mod N+1 happen to agree, which for going from 4 to 5 shards is only about one key in five. So roughly 80% of the data has to be copied across the network, while the cluster is under the load that prompted the change. Consistent hashing or a fixed number of virtual partitions assigned to nodes limits movement to about 1/N of the data.

Consistent hashing with virtual nodes

flowchart LR subgraph Ring[Hash ring, clockwise] A1[Node A vnode 1] --> B1[Node B vnode 1] B1 --> C1[Node C vnode 1] C1 --> A2[Node A vnode 2] A2 --> B2[Node B vnode 2] B2 --> C2[Node C vnode 2] C2 --> A1 end K1[Key hash lands between A1 and B1] -. owned by .-> B1

With one point per node, arcs are uneven and a node's failure dumps its whole arc onto one neighbour. Virtual nodes give each physical node many points (100–256), so load evens out, a failed node's keys scatter across all survivors, and a bigger machine can simply take more points. Replicas go to the next distinct physical nodes clockwise.

Q: What problem do virtual nodes solve in consistent hashing?

A: Uneven load and uneven failover. With one ring position per server, random placement leaves some servers owning much larger arcs than others, and when a server dies its entire arc moves to its single clockwise neighbour, possibly overloading it. Giving each server many virtual positions averages the arcs out, spreads a failed server's keys across all the remaining servers, and lets heterogeneous hardware take a proportional number of positions.

Choosing a shard key

A good shard key:

  • appears in the most frequent queries, so they route to one shard;
  • has high cardinality, so data can spread across many shards;
  • has even load — no single value carries a large fraction of traffic;
  • keeps data that is used together on the same shard (a user's orders with the user; a tenant's rows with the tenant).

Common choices: user id for consumer apps, tenant id for B2B SaaS, order id for order processing, device id plus time bucket for telemetry. Bad choices: country, status, created date alone, or anything with a few dominant values.

Q: What makes a good shard key for a B2B SaaS application, and what is its weakness?

A: Tenant id is usually the best choice: nearly every query is already scoped to one tenant, so it routes to a single shard; a tenant's data stays together, so joins and transactions remain local; and moving or isolating a tenant is a shard-level operation. The weakness is skew — one enterprise tenant can be thousands of times larger than the median, overloading its shard. Mitigations: place large tenants on dedicated shards via a directory, or use a compound key (tenant id plus a sub-key) for the few tenants that exceed a single shard.

Hot keys and hot partitions

Even with a good key, a single key can be hot: a celebrity's profile, a viral product, one giant tenant. Options, cheapest first:

  1. Cache the hot key's reads in front of the shard (and in-process for extreme cases).
  2. Replicate hot read-mostly keys to several shards and pick one at random.
  3. Split the key for writes: append a suffix 0–N (counter:123:7), write to a random suffix, and sum on read.
  4. Move the hot tenant or key to its own shard via a directory.
Q: A single product's like counter receives 50,000 increments per second and saturates its shard. How do you fix it?

A: Split the counter. Store it as N sub-counters (for example 50 keys, likes:product:0 to likes:product:49) that hash to different shards; each increment goes to a random sub-counter, spreading writes across shards, and a read sums the N values — cached for a second or two, since exact real-time accuracy is not needed. Alternatively, buffer increments in memory or a stream and apply them in batches. The price is a slightly more expensive and briefly stale read.

Resharding without downtime

sequenceDiagram participant App participant Old as Old shard participant New as New shard participant Dir as Routing map App->>Old: Normal reads and writes Note over Old,New: 1. Copy snapshot of moving range to new shard Old->>New: Bulk copy plus change stream catch-up App->>Old: Writes continue Old->>New: Change stream keeps new shard current Note over App,Dir: 2. Brief write pause or dual write for the range App->>Dir: Flip routing for the range to new shard App->>New: Reads and writes for the moved range Note over Old: 3. Verify, then delete moved data from old shard

The pattern is always copy, catch up, cut over, clean up: snapshot the data to move, stream changes made during the copy, pause writes for that range for seconds (or dual-write), flip the routing entry, verify, and only then delete. Move small units (a virtual partition or a tenant), not whole shards, and throttle the copy so it does not starve production traffic.

Q: What are the steps of an online shard migration, and where is the risk?

A: Copy a snapshot of the moving range to the destination, catch up with a change stream (CDC or the replication log) until the destination lags by seconds, briefly block or queue writes for that range while the last changes apply, flip the routing map so clients go to the destination, then verify counts and checksums and delete from the source after a safety period. The risk concentrates at the cutover: a write accepted by the source after the final catch-up but before the routing flip is lost unless writes are fenced, and clients with a cached old routing map keep writing to the source — so the source must reject writes for ranges it no longer owns.

Cross-shard queries and secondary indexes

  • Scatter-gather: send the query to every shard and merge. Latency is set by the slowest shard, and load grows with shard count. Fine for rare admin queries, bad for hot paths.
  • Local secondary indexes (document-partitioned): each shard indexes only its own data; a query by the secondary attribute must scatter.
  • Global secondary indexes (term-partitioned): the index itself is partitioned by the indexed value; a lookup hits one index shard, but every write updates the index asynchronously on another shard, so the index is eventually consistent.
  • Denormalise: keep a second copy of the data keyed the other way (orders by customer and orders by merchant).
  • Cross-shard transactions: two-phase commit or a distributed SQL database (Spanner, CockroachDB, Vitess with 2PC); correct but slower and more failure-prone. Design keys to make them rare.
Q: What is the difference between a local and a global secondary index in a sharded store?

A: A local index lives on each shard and covers only that shard's rows, so it is updated in the same local write and is consistent, but a query by the indexed attribute must ask every shard (scatter-gather). A global index is itself partitioned by the indexed value, so a query goes to one index partition, but each base write must also update an index entry that may live on another shard — usually asynchronously, so the index can briefly lag the data. DynamoDB exposes exactly this choice as LSI versus GSI.

Q: How do you enforce a globally unique email address when users are sharded by user id?

A: The uniqueness check needs a single place keyed by email. Keep a separate lookup table sharded by email (email to user id) and claim the email there with a conditional insert before, or transactionally with, creating the user; if the claim fails the email is taken. If user creation fails after the claim, release it or let a cleanup job reconcile orphaned claims. Scatter-checking every user shard is racy — two sign-ups can both see no match — so it does not enforce uniqueness.

Unique ids across shards

Auto-increment ids per shard collide. Options: UUIDv4 (random, index-unfriendly), UUIDv7 or ULID (time-ordered and random, good default), Snowflake-style 64-bit ids (timestamp + worker id + sequence), or a central id range allocator handing out blocks. Time-ordered ids keep B-tree inserts local and make ids roughly sortable.

Q: Why are time-ordered ids such as UUIDv7 or Snowflake ids preferred over random UUIDv4 as primary keys?

A: Random UUIDs insert at random positions in the primary key B-tree, causing page splits, poor cache locality and write amplification as the index grows beyond memory. Time-ordered ids append near the end of the index like an auto-increment, keeping the hot pages in cache, while still being generated without coordination on any node. They are also roughly sortable by creation time. The trade-off is that they leak approximate creation time, and in a range-partitioned store they concentrate new writes, so they suit hash-partitioned or single-node indexes best.

See also