Step 06 of 22
Partitioning & sharding
Split data across machines: choose a partition key, compare range and hash partitioning, rebalance without moving everything, and handle hot keys, secondary indexes and cross-shard queries.
Replication copies the same data onto several machines. It scales reads but not writes, and every replica still has to hold the whole dataset. When the write rate or the data size outgrows the largest machine you can reasonably buy, you partition: split the data so each machine holds and serves only a slice. (Databases call the slices partitions, shards, regions, tablets or vnodes; they all mean the same thing.) Partitioning is powerful and expensive. This lesson covers the decisions that determine whether it goes well: the key, the scheme, rebalancing, hot spots, and queries that span shards.
The partition key decides everything
Every record gets a partition key, and a function maps keys to partitions. Choose the key from your access patterns, because the goal is for the common queries to touch one partition:
| Workload | Good key | Why |
|---|---|---|
| Multi-tenant SaaS | tenant_id | Almost every query is scoped to one tenant |
| Chat | conversation_id | Messages are read and written per conversation |
| Social graph | user_id | "My followers", "my timeline" are per user |
| Time-series metrics | (metric_id, time bucket) | Spreads writes while keeping a series' recent data together |
A key that queries don't use forces every query to ask every partition — the worst outcome of partitioning. And a key that sends most traffic to a few values creates hot partitions no matter how many machines you add.
Range partitioning
Assign each partition a contiguous range of keys: A–F on shard 1, G–M on shard 2, and so on. HBase, Bigtable, CockroachDB, TiDB and MongoDB's ranged sharding work this way.
- Range queries are efficient. "All orders for customer 4,200–4,300" or "events from 10:00 to 10:05" touch one or two partitions, and keys within a partition are stored sorted.
- Boundaries adapt to data. When a partition grows too large, split it in two; when two shrink, merge them.
- Sequential keys create a hot spot. If the key is a timestamp or an auto-increment ID, every new write lands in the last range, so one partition takes all the writes while the rest sit idle. Prefix the key with something that spreads writes (
sensor_idbefore the timestamp) or use hash partitioning.
Hash partitioning
Hash the key and assign partitions by hash value. A good hash scatters similar keys uniformly, so load spreads evenly even when the keys themselves are sequential. Cassandra, DynamoDB and Redis Cluster all hash.
The cost: range queries are gone. Adjacent keys land on different partitions, so "orders between two dates" must ask all of them. A common compromise is a compound key: hash the first part to choose the partition, and sort by the second part within it. In Cassandra, PRIMARY KEY ((user_id), created_at) puts all of a user's rows on one partition, sorted by time, so "this user's latest 20 posts" is a single sequential read. See range vs hash partitioning.
Rebalancing without moving everything
Partitions must move when you add or remove machines. The naive assignment, partition = hash(key) mod N, is a disaster here: change N from 4 to 5 and about 80% of keys map to a different node, so nearly the whole dataset moves at once.
Two designs fix it.
Consistent hashing puts nodes and keys on a ring; each key belongs to the next node clockwise. Adding a node takes keys only from its neighbour, so about 1/N of the data moves. Virtual nodes (each machine placed on the ring many times) even out the shares. The DevSecOps load-balancing lesson works through the ring step by step; try it here. See consistent hashing.
Replace 'backend' with 'shard'. Under modulo hashing, removing one node remaps most keys; on the ring only the removed node's keys move.
A fixed number of partitions is simpler and more common in practice. Create many more partitions than nodes up front — Redis Cluster has 16,384 hash slots, Elasticsearch indexes have a fixed shard count, Kafka topics a fixed partition count — and assign several partitions to each node. Adding a node means moving whole partitions to it; the key-to-partition mapping never changes, only the partition-to-node table does.
slot = CRC16(key) mod 16384 # Redis Cluster
node = slot_table[slot] # e.g. slots 0–5460 → A, 5461–10922 → B, 10923–16383 → CChoose the partition count for your largest future size: too few and each partition becomes huge and slow to move; too many and per-partition overhead (files, metadata, open connections) adds up.
Routing requests to the right partition
Something has to know which node holds which partition. Three common designs:
- Any node routes. The client talks to any node, which forwards to the owner (Cassandra coordinators).
- A routing tier. A proxy layer knows the map and forwards (MongoDB's
mongos, Vitess's VTGate). - Smart clients. The client library caches the map and talks to the owner directly; a node that receives a misdirected request replies with a redirect (Redis Cluster's
MOVED).
The authoritative map usually lives in a small consensus-backed store — ZooKeeper, etcd, or the database's own metadata service — so every router agrees on it. See consensus and Raft.
Hot partitions and hot keys
Even with perfect hashing, one key can be hot: a celebrity's profile, a viral post, a single big customer in a multi-tenant system. All requests for that key go to one partition, and adding machines does nothing. See hot partitions.
Defences, from lightest to heaviest:
- Cache it. Hot reads are usually the easiest case; a cache (even an in-process one) absorbs them.
- Salt the key for writes. Write to
post:42#0…post:42#9, chosen at random, so ten partitions share the load. Reads must fetch all ten and combine them — fine for counters (sum them), awkward for anything else. - Give big tenants their own partition or even their own cluster. Many SaaS systems move their largest customers to dedicated shards.
- Split by a second key for the hot entity (for a huge conversation, partition by
(conversation_id, month)).
Detect hot keys before users do: track per-partition request rates and alert when one is far above the median.
Secondary indexes across partitions
Partitioning by user_id makes "this user's orders" fast. But how do you find "all orders with status failed"? Status isn't the partition key. Two options, each expensive in a different place. See secondary indexes on sharded data.
Local (document-partitioned) index. Each partition indexes only its own rows. Writes are cheap — one partition — but a query by status must ask every partition and merge the results (scatter-gather). Latency is set by the slowest partition, and the query load multiplies by the partition count.
Global (term-partitioned) index. The index itself is partitioned by the indexed value: all failed entries on one index partition. Reads hit one index partition, but each write must update the data partition and one or more index partitions, usually asynchronously — so the index lags the data. DynamoDB's global secondary indexes work this way and are eventually consistent.
If a query isn't on the hot path, scatter-gather over a local index is often fine. For hot queries on another key, a global index — or a second copy of the data partitioned by that key — is the usual answer.
Cross-partition queries and transactions
Joins and transactions that span partitions are where partitioned systems get hard:
- Design so they're rare. Choose the key so related data lives together (a tenant's users, projects and tasks all keyed by
tenant_id). - Denormalise data that's read together, accepting that copies must be kept in sync (normalization vs denormalization).
- Use a database that does distributed transactions (Spanner, CockroachDB, YugabyteDB) when you genuinely need them, and accept the extra latency of a commit protocol across nodes.
- Use sagas for multi-step business operations across services, with compensating actions instead of a single atomic commit (sagas).
Before any of this, check whether you need partitioning at all. A single PostgreSQL primary on a large machine handles terabytes and thousands of writes per second; partitioning a database that would have fit on one machine adds every problem in this lesson and solves none.
Practice
You're partitioning the database of a multi-tenant invoicing product: 40,000 tenants, but the largest 20 account for 35% of all rows and traffic. Queries: list a tenant's invoices by date; fetch one invoice by ID; a nightly job finds every overdue invoice across all tenants; and support staff search invoices by customer email across tenants.
Choose a partition key and scheme, then explain how you'd handle the big tenants, the overdue job, and the support search.
Show solution
Key and scheme. Partition by tenant_id, because nearly every query is scoped to one tenant and keeping a tenant's data together keeps those queries — and transactions — single-partition. Use hash partitioning over a fixed number of partitions (for example 1,024 logical shards mapped to a handful of physical nodes) so tenants spread evenly and nodes can be added by moving whole shards. Within a shard, index (tenant_id, issued_at) for "invoices by date", and make invoice IDs carry or map to the tenant (for example a (tenant_id, invoice_id) primary key, with the tenant in the URL) so fetching by ID is a single-shard lookup.
Big tenants. The top 20 would make their shards hot. Give each of them a dedicated shard (or a dedicated node) by recording an override in the shard map, and move them there gradually. If a single tenant ever outgrows a node, sub-partition it by (tenant_id, year).
Overdue job. It's a cross-tenant batch, not a user request, so scatter-gather is acceptable: run it per shard in parallel, each using a local partial index such as ON invoices (due_date) WHERE status = 'unpaid', and throttle it so it doesn't compete with daytime traffic.
Support search by email. Scatter-gathering every shard on each search is slow and loads every node. Maintain a global lookup instead: a separate table or search index partitioned by email (email → [(tenant_id, invoice_id)]), updated asynchronously from invoice changes (for example through an outbox and a consumer). It lags by seconds, which support can live with.
Key takeaways
- Partition when writes or data size outgrow one machine; replicas scale only reads.
- Pick the partition key from access patterns so common queries touch one partition.
- Range partitioning keeps range scans fast but makes sequential keys hot; hash partitioning spreads load but scatters ranges. Compound keys combine them.
- Never use
hash mod Nfor placement: use consistent hashing or many fixed partitions moved whole. - Hot keys need caching, salting or dedicated partitions; more machines don't help.
- Secondary indexes are either local (cheap writes, scatter-gather reads) or global (cheap reads, slower and lagging writes).
Check yourself
Quiz
0/4 correct · 4 left
1.An events table is range-partitioned by event timestamp. What goes wrong under heavy ingestion?
2.Why do Redis Cluster and Kafka use a fixed number of partitions (slots) much larger than the node count?
3.Orders are partitioned by user_id with a local index on status. What does a query for all orders with status = 'failed' cost?
4.One celebrity's like counter gets 200,000 increments per second, overwhelming its partition. Which fix addresses it?
Concepts in this lesson
One-page notes on each idea, for review or a quick lookup later.
- Range vs hash partitioningTwo ways to decide which shard owns a key, and what each does to range queries and hot spots.
- Consistent hashingPlace nodes and keys on a ring so adding a node moves about 1/N of the keys, not all of them.
- Hot partitions and hot keysWhen one key gets most of the traffic: detection, key salting, splitting and caching.
- Secondary indexes on sharded dataLocal indexes mean scatter-gather reads; global indexes mean distributed writes.
Finished reading?
Mark it complete to fill in your roadmap.