Sharding: shard keys, consistent hashing and moving data live

Sharding splits one dataset across several machines so that no single machine has to hold all of it or absorb all of its writes. It is how every large database scales past one box, and it is also the most expensive architectural decision most teams make: easy to start, hard to change, and shaping every query that comes after.

This article covers when sharding is worth it, how to choose a shard key, range against hash partitioning, why naive hashing makes growth painful and how consistent hashing and fixed partitions fix it, and the problems that remain: hot keys, cross-shard queries, and moving data while the system is live.

When to shard, and when not yet

Sharding solves two problems: data that no longer fits on one machine, and write throughput that one machine can’t sustain. Read throughput alone is not a reason to shard. Read replicas solve that more cheaply, as do caches.

Before sharding, exhaust the cheaper options. A single modern database server with hundreds of gigabytes of memory and fast NVMe storage handles more than most products will ever need. Archive cold data. Move analytics to a separate store. Split unrelated tables onto separate databases, which is sometimes called vertical partitioning and keeps every query single-node.

Then shard knowing what it costs:

• Queries that don’t include the shard key have to ask every shard.
• Transactions that span shards need a distributed protocol, or have to be designed away.
• Joins across shards move into application code.
• Growing the cluster means moving data while serving traffic.
• Every operational task, from backups and schema changes to failovers, now happens N times.

Choosing the shard key

The shard key decides which shard each row lives on, and it is very hard to change later. A good key has three properties:

• Many distinct values, spread evenly. Sharding by country puts a third of your users on the shard holding the largest country. Sharding by user ID spreads them out.
• Present in most queries. A query that names the key goes to one shard; a query that doesn’t goes to all of them. Look at your actual query log, not the data model.
• Groups data that changes together. If transactions touch one tenant’s rows, shard by tenant, and they stay single-shard. For multi-tenant SaaS, tenant_id is usually the right key for exactly this reason.

The classic mistake is a key that grows over time, such as a timestamp or an auto-increment ID, combined with range partitioning. Every new write has the largest key, so every new write lands on the last shard, and one machine takes the entire write load while the others sit idle.

Range or hash

Range partitioning gives each shard a contiguous range of keys: A–F, G–M, and so on. Keys stay sorted, so range scans (“all orders for this customer in March”) touch one shard or a few neighbouring ones. Bigtable, HBase, Spanner, CockroachDB and TiKV all work this way. They split a range automatically when it grows too big or too busy, and move ranges between nodes to balance load. The risk is the hot spot described above: a monotonically increasing key keeps writing to the same range.

Hash partitioning hashes the key and assigns shards by hash. Neighbouring keys scatter, so load spreads evenly even when keys are sequential, but range scans on the key now touch every shard.

Many systems combine the two. In Cassandra and DynamoDB the primary key has two parts: a partition key that is hashed to pick the node, and a sort key (Cassandra’s clustering columns) that keeps rows sorted within the partition. With (customer_id, order_date), one customer’s orders live together and in date order, while customers spread across the cluster.

Why hash mod N breaks

The obvious hash scheme is shard = hash(key) mod N. It spreads keys evenly and needs no state. The problem appears when N changes. A key stays where it is only if hash mod N = hash mod (N + 1), which holds for about 1 in N + 1 keys. Going from 4 shards to 5 moves about 80% of the data, when the minimum needed to fill the new shard is 20%.

Moving 80% of the data is a migration, not a scale-up. For a cache it means the hit rate collapses to about 20% the moment you add a node, which is exactly when you needed more capacity.

Consistent hashing

Consistent hashing (Karger et al., 1997) hashes both keys and shards onto the same circle of values. Each key belongs to the first shard clockwise from it. Adding a shard places it at some point on the ring, and it takes over only the keys between it and its predecessor; every other key stays where it is. Removing a shard hands its keys to its successor.

With one point per shard, the arcs are very uneven, and when a shard leaves, its whole load falls on a single neighbour. The fix is virtual nodes: place each shard at many points, often 100–256. Each shard then owns many small arcs scattered round the ring, its share of keys evens out, and when it leaves, its load spreads across all the others. Virtual nodes also let you give a bigger machine more points. Amazon’s Dynamo (DeCandia et al., 2007) popularised the approach, and Cassandra and Riak inherited it.

Two alternatives come up often:

• Rendezvous hashing (highest random weight; Thaler & Ravishankar, 1998). For each key, score every shard with hash(key, shard) and pick the highest. It needs no ring and balances well, but costs O(N) per lookup.
• Jump consistent hash (Lamping & Veach, 2014). A few lines of arithmetic that map a key to one of N numbered buckets, with no memory, near-perfect balance and the minimum possible movement. The catch is that buckets are numbered: you can add or remove only the last one, so it suits shards that are themselves replicated, rather than individual machines that fail.

Fixed partitions and slots

Most production systems take a simpler route: create far more partitions than nodes, up front, and never change the number. Keys map to partitions with a fixed hash; partitions map to nodes through a table that can change. Growing the cluster means moving whole partitions to the new node, and no key ever changes partition.

• Redis Cluster has 16,384 hash slots: slot = CRC16(key) mod 16384. Each node owns a set of slots, and resharding migrates slots one at a time. Hash tags let you co-locate related keys: only the part in braces is hashed, so {user:42}:cart and {user:42}:profile land on the same slot and can be used in one transaction.
• Kafka fixes the partition count per topic. Adding partitions later is possible, but it changes which partition a key maps to, which breaks per-key ordering, so in practice teams create enough at the start (see the Kafka article).
• Elasticsearch fixes the number of primary shards when an index is created; changing it means splitting or shrinking by a whole factor, or reindexing.

The trade-off is choosing the number up front. Too few, and you can’t spread across more nodes later. Too many, and each partition carries fixed overhead in memory, file handles and metadata. A common rule is to aim for ten or more partitions per node at the largest size you expect.

Routing requests

Something has to know which node owns which partition. There are three common places to put that knowledge:

• In the client. Smart clients cache the partition map and send each request straight to the right node. Redis Cluster clients do this, and a node that no longer owns a slot answers MOVED so the client can refresh its map. It is the fastest option, but every client library must implement it correctly.
• In a proxy tier. Clients talk to a stateless router that forwards each query. Vitess’s VTGate does this for sharded MySQL, so applications can mostly treat the cluster as one database. It costs an extra hop.
• Any node forwards. In Cassandra, any node can coordinate a request and forwards it to the replicas that own the key, learning the ring through gossip.

Wherever it lives, the map itself needs a source of truth that every router agrees on. That is usually a consensus-backed store such as ZooKeeper or etcd, or the database’s own metadata ranges.

Hot keys and skew

Hashing spreads keys evenly, not load. If one key gets a large share of the traffic, its shard gets that share, whatever the hash function does. Examples are a celebrity’s account, a viral post, or the one enterprise tenant that is a hundred times larger than the rest.

• Split the key. Append a suffix from 0 to k − 1 to a hot key, so its writes spread over k shards, and read all k pieces and combine them. Counters and append-only feeds suit this well.
• Cache hot reads in front of the shard. Read-heavy hot keys usually belong in a cache, even a small in-process one.
• Give big tenants their own shard. With a directory that maps tenants to shards, rather than a hash, you can move a large tenant onto dedicated hardware without touching anyone else.
• Let the database split for load. DynamoDB, CockroachDB and Spanner split partitions that are hot, not just ones that are big, but a single key can never be split below one partition.

Cross-shard queries and transactions

A query that filters on anything other than the shard key has to be sent to every shard, and its latency is the latency of the slowest one. With many shards, that is a tail-latency problem: if each shard is slow 1% of the time, a query across 100 shards is slow on most requests (the same arithmetic as in the microservices article).

Secondary indexes come in two forms, each with a cost:

• Local indexes (document-partitioned). Each shard indexes only its own rows. Writes stay single-shard, but every index lookup is a scatter-gather.
• Global indexes (term-partitioned). The index is itself sharded by the indexed value, so a lookup goes to one index shard. But a single write now updates rows on several shards, so global indexes are usually updated asynchronously and can be briefly stale. DynamoDB’s global secondary indexes work this way.

Transactions across shards need two-phase commit or a consensus protocol, as Spanner and CockroachDB provide. They work, but cost extra round trips and hold locks across the network. The more durable fix is in the data model: choose the shard key so that the operations that need atomicity touch one shard, and handle the rare cross-shard workflow with a saga.

Resharding a live system

Eventually you will need to move data: more shards, a new key, or a hot tenant to isolate. Doing it without downtime follows the same pattern whatever the database:

1. Copy. Bulk-copy the data to its new location while the old one keeps serving.
2. Catch up. Stream changes made since the copy started, from the database’s change log, until the new location is only seconds behind.
3. Verify. Compare row counts and checksums between old and new.
4. Switch reads, then switch writes, with a short write pause or a per-key cut-over to make the last changes safe.
5. Keep the reverse stream running for a while, so you can switch back if something is wrong.

Vitess automates exactly this with VReplication, and Redis Cluster and CockroachDB do the equivalent per slot or per range. Whatever tool you use, throttle the data movement: copying terabytes competes with production traffic for disk and network, and an unthrottled resharding job can cause the outage it was meant to prevent.

A consistent-hash ring

A compact ring with virtual nodes, binary-search lookup, and Dynamo-style replica placement that walks clockwise to collect the next n distinct shards.

// A consistent-hash ring with virtual nodes. Each shard owns the keys between each of
// its virtual nodes and the one before it, so a new shard only takes keys from its
// neighbours on the ring, never shuffles keys between existing shards.
import { createHash } from "node:crypto";

// Any well-mixed hash works; it must be identical in every client.
const point = (s: string) => createHash("md5").update(s).digest().readUInt32BE(0);

export class Ring {
  private points: { at: number; shard: string }[] = [];

  constructor(shards: string[], private vnodes = 128) {
    shards.forEach((s) => this.add(s));
  }

  add(shard: string) {
    for (let v = 0; v < this.vnodes; v++) this.points.push({ at: point(shard + "#" + v), shard });
    this.points.sort((a, b) => a.at - b.at);
  }

  remove(shard: string) {
    this.points = this.points.filter((p) => p.shard !== shard);
  }

  // The first virtual node at or after the key's position, wrapping round to the start.
  owner(key: string): string {
    return this.points[this.first(point(key)) % this.points.length].shard;
  }

  // Replica placement as in Dynamo: walk clockwise, take the next n distinct shards.
  preferenceList(key: string, n: number): string[] {
    const start = this.first(point(key)), out: string[] = [];
    for (let k = 0; k < this.points.length && out.length < n; k++) {
      const s = this.points[(start + k) % this.points.length].shard;
      if (!out.includes(s)) out.push(s);
    }
    return out;
  }

  private first(h: number): number {
    let lo = 0, hi = this.points.length; // binary search for the first point >= h
    while (lo < hi) { const mid = (lo + hi) >> 1; if (this.points[mid].at < h) lo = mid + 1; else hi = mid; }
    return lo;
  }
}

References

1. Karger, D. et al. (1997). Consistent Hashing and Random Trees: Distributed Caching Protocols for Relieving Hot Spots on the World Wide Web. STOC. doi:10.1145/258533.258660
2. DeCandia, G. et al. (2007). Dynamo: Amazon’s Highly Available Key-value Store. SOSP. doi:10.1145/1294261.1294281
3. Lamping, J. & Veach, E. (2014). A Fast, Minimal Memory, Consistent Hash Algorithm. arXiv:1406.2294.
4. Thaler, D. G. & Ravishankar, C. V. (1998). Using Name-Based Mappings to Increase Hit Rates. IEEE/ACM Transactions on Networking. doi:10.1109/90.663936
5. Chang, F. et al. (2006). Bigtable: A Distributed Storage System for Structured Data. OSDI. usenix.org
6. Corbett, J. C. et al. (2012). Spanner: Google’s Globally-Distributed Database. OSDI. usenix.org
7. Redis. Redis Cluster Specification. redis.io
8. Vitess. Sharding and VReplication documentation. vitess.io/docs
9. Kleppmann, M. (2017). Designing Data-Intensive Applications, ch. 6: Partitioning. O’Reilly. dataintensive.net