all_lessons / data_intensive_systems / 11 · partitioning lesson 12 / 35 · ~17 min

Part 5 · Distribution

Partitioning: Hash, Range, Hot Keys, and Rebalancing

Part 4 spent its energy on redundancy — keeping copies of the same data consistent, from one leader with followers (lesson 08), to many writers reconciled by a quorum (lesson 09), to a device holding its own primary copy offline (lesson 10). But redundancy does nothing for a dataset that no longer fits on one machine, or a write rate that swamps a single node's disk and CPU: every replica still holds the whole thing. This part — distribution — makes the orthogonal move: instead of copying the data, we split the data itself. Partitioning (sharding) divides the dataset across machines so each node owns only a slice. The instant you do that, one decision dominates everything — how you choose which key goes where. That choice, the shard key, is a performance contract: it fixes locality, balance, and which queries are cheap point lookups versus expensive fan-outs.

DDIA source
Arc drawn from Martin Kleppmann, Designing Data-Intensive Applications, Chapter 6 (Partitioning) — partitioning by key range vs hash of key, skew and hot spots, partitioned secondary indexes, rebalancing strategies, and request routing. Original synthesis; the feature-store and vector-index framings are ours.
Linear position
Prerequisite: Lessons 08–10 — replication keeps copies of data consistent (single-leader lag and failover; leaderless quorums w + r > n; local-first sync). We build on "a node can hold and serve data" and treat replication as orthogonal: in real systems each partition is itself a small replica set.
New capability: Choose a shard key on purpose; predict which queries become scatter-gather; quantify hot-key skew and mitigate it; pick a secondary-index layout; and rebalance a cluster as nodes come and go without moving almost all the data.
The plan
Five moves. (1) Why partition at all, and the one rule that makes it hard: skew. (2) The two ways to assign keys — by key range (range scans, hot-range risk) vs by hash of key (even spread, no range scans) — and why the shard key is a performance contract that turns some queries into scatter-gather. (3) Hot keys and skew, with a worked number for what one celebrity key does to a partition. (4) A partition-skew widget: dial the hot-key share and partition count, watch the load bars and the max-over-average imbalance. (5) Secondary indexes (local vs global) and rebalancing (why "hash mod N" is a disaster, fixed partitions, consistent hashing, dynamic partitioning, and how a client finds the right partition).

1 · Why partition, and the one rule that makes it hard

A single machine has finite disk, RAM, CPU, and network bandwidth. Replication (lessons 08–10) copies the whole dataset onto several machines for availability and read throughput — but every copy is still the whole thing, so it does nothing when the dataset itself exceeds one disk or the write rate exceeds one node's CPU. Partitioning (also called sharding) is the orthogonal answer: split the dataset into partitions (shards) so that each piece of data lives on exactly one partition, and different machines own different partitions. Capacity and load now scale with the number of machines.

The two combine. Each partition is normally also replicated — a partition is a small replica set running the leader or quorum protocol from 08/09. So a record is identified by "which partition" (this lesson) and then "which replicas of that partition" (last lesson). Keep the two axes separate in your head: partitioning is about where a key lives; replication is about how many copies of that location exist.

The goal sounds trivial: spread data and requests evenly. The catch is that even bytes and even load are different things, and real workloads are skewed — a few keys get most of the traffic. A partition can hold an average number of bytes yet receive celebrity-level request volume; another can be huge but cold. Partitioning would be easy if every key were equally popular and every query touched exactly one key. It is hard precisely because data and attention are skewed, and a single overloaded partition — a hot spot — caps the whole cluster's throughput no matter how many nodes you add.

2 · Two ways to assign keys: range vs hash

You need a function from key to partition. There are two canonical choices, and they trade exactly opposite things.

Partition by key range. Assign each partition a contiguous, sorted range of keys — partition 1 owns keys "A" through "F", partition 2 owns "G" through "M", and so on (like the volumes of a paper encyclopedia). Because keys stay sorted, a range scan — "give me every key between X and Y" — touches only the few partitions covering that interval, which is the great win. The danger is built into the same property: if the access pattern concentrates on one end of the key order, that one partition becomes a hot spot. The classic trap is keying by timestamp: every new write has "now" as its key, so today's partition takes 100 percent of writes while every other partition sits idle. Sequential auto-increment ids have the same failure: the tail shard burns.

Partition by hash of key. Run the key through a hash function (a deterministic scramble producing a uniform-looking number) and assign each partition a range of hash values. A good hash destroys the input's structure, so even sequential keys land on different partitions and load spreads evenly — this is the great win, and it tames the timestamp hot spot automatically. The price is that you lose sorted order: adjacent keys are deliberately scattered, so a range scan now has no contiguous home and must query every partition. You traded range locality for balance.

key --[ partition function ]--> partition (shard) RANGE: "alice"|"bob" -> P0 "carol"|"dave" -> P1 "eve"|"frank" -> P2 + range scan "c..e" hits only {P1,P2} - hot end: keying by timestamp -> ALL writes hit the newest partition HASH: h("alice")=8a.. -> P2 h("bob")=12.. -> P0 h("carol")=c4.. -> P1 + even spread; sequential/timestamp keys no longer pile up - range scan "c..e" has no home -> must hit {P0,P1,P2} (scatter-gather)

The shard key is a performance contract. The column (or composite of columns) you partition on decides three things at once, for the life of the table:

Compound keys split the difference. Use a composite shard key: hash the first part for balance, keep the rest sorted for local range scans. Partition by (tenant_id, timestamp) — hash on tenant_id so tenants spread evenly across the cluster, but keep each tenant's rows sorted by timestamp within its partition. Now "latest 50 events for tenant X" is a local range scan on one partition, while no single tenant's clock creates a global hot end. The cost: a query that omits tenant_id (e.g., "all events across all tenants in the last hour") is back to scatter-gather.

3 · Hot keys and skew: the number that bites

Hashing fixes skew from structured keys (timestamps, sequences). It does not fix skew from a single hot key — a celebrity user, a viral post, a single hugely popular feature entity. A hot key has exactly one shard-key value, so the hash sends 100 percent of its traffic to exactly one partition. The hash spread out the other keys; it cannot spread out one key's requests, because by definition they all share the same key.

Worked number — what one hot key does
Suppose you have n = 10 partitions and a perfectly uniform hash, so the ideal per-partition share of load is 1 / 10 = 10%. Total cluster traffic is 100,000 requests per second (rps). Now one celebrity account draws 30% of all reads.

That key lives on a single partition, call it P7. The remaining 70% of traffic spreads evenly over all 10 partitions (the celebrity's key is one of many on P7, but its 30 percent dominates). So:
  • Each ordinary partition's load = 70% / 10 = 7% = 7,000 rps.
  • P7's load = 7% + 30% = 37% = 37,000 rps.
  • Cluster average = 100% / 10 = 10% = 10,000 rps.

The imbalance ratio (hottest partition over average) is 37,000 / 10,000 = 3.7×. P7 must be provisioned for 37,000 rps while nine machines idle at 7,000 — you are paying for 10 machines and getting the throughput ceiling of roughly 100,000 / 3.7 ≈ 27,000 rps of useful uniform capacity before P7 saturates. Adding more partitions does not help: the celebrity's 30 percent still lands on one partition.

Mitigations — you must break the single key apart:

4 · Feel the knobs: the partition-skew widget

Set the number of partitions and the share of traffic taken by a single hot key. The widget draws each partition's load as a bar (the hot key's contribution stacked on top of its uniform baseline) and reports the cluster average, the hottest partition, and the imbalance ratio (max over average) — the number that decides your real throughput ceiling.

Partition skew — what a hot key does to balance
Each bar is one partition's share of total load. The pale segment is its slice of the uniformly-spread traffic; the dark segment on the hot partition is the single hot key. A flat row of bars is ideal; one tall bar is a hot spot that caps the whole cluster.
Average load / partition
Hottest partition
Imbalance (max / avg)
Useful uniform ceiling
Show the core JS
// hot share H (fraction) is split into k sub-keys on k distinct partitions.
// the remaining (1 - H) spreads uniformly over all n partitions.
const base = (1 - H) / n;              // every partition gets this
const hotPer = H / k;                  // each of the k hot sub-keys adds this
// hottest = a partition that carries one hot sub-key:
const maxLoad = base + (k <= n ? hotPer : H / n);
const avg = 1 / n;                     // ideal share
const imbalance = maxLoad / avg;       // throughput ceiling = 1 / imbalance

Push the hot-key share up with k = 1 and watch one bar tower over the rest; then raise k (splitting the hot key into sub-keys) and watch that tower flatten back toward the average — that is the random-suffix mitigation from §3, made visible.

5 · Secondary indexes and rebalancing

So far a query routes by the primary shard key. But applications also search by other fields — "all red cars", "all users in region EU". A secondary index is a lookup structure on a non-key field. Partitioning one is its own design choice with two options that, again, trade opposite costs.

Local (document-partitioned) index
Each partition indexes only its own rows. A write updates the index on the same partition that holds the row — writes stay local, one partition touched. But a query by the secondary field doesn't know which partitions hold matches, so a read scatter-gathers: ask every partition, merge. Cheap writes, expensive reads. This is the common default (Elasticsearch, MongoDB).
Global (term-partitioned) index
The index itself is partitioned by the indexed term — "all red cars" lives on whichever partition owns the term "red", wherever the cars physically are. A read by the secondary field goes to one partition (targeted, fast). But a single row write may touch terms living on several partitions, so the write scatters and the index update is usually asynchronous (it lags the primary). Expensive writes, cheap reads.

The mirror image is exact: local = scatter-gather reads, local writes; global = scatter writes, targeted reads. Pick by which side of the workload is hot.

Rebalancing: moving partitions as nodes come and go

Rebalancing is moving partitions between nodes when you add machines (to grow) or lose them (failure, decommission). The whole art is moving the minimum data, because every moved byte costs network and disk bandwidth that competes with live traffic.

Why "hash mod N" is a disaster
The naive scheme: partition = hash(key) mod N, where N is the number of nodes. It looks balanced — and it is, until N changes. Go from N = 10 to N = 11 and hash(key) mod 11 gives almost every key a different answer than hash(key) mod 10 did. Concretely, only about 1 / 11 ≈ 9% of keys keep their home; roughly 91% of the entire dataset must move across the network just to add one node. The number of nodes must never appear in the placement function.

The fixes all decouple the placement function from the node count:

Request routing. Once partitions move around, a client needs to find the partition that owns its key. This is service discovery for data. Three shapes: (1) clients contact any node, which forwards to the right one (gossip-style, e.g. Cassandra); (2) a routing tier sits in front and forwards (e.g. a mongos router); (3) clients consult a coordination service that holds the authoritative partition-to-node map. That coordination service — typically ZooKeeper or etcd — is itself a consensus system, which is exactly the machinery lesson 17 builds. Whoever holds the map is the source of truth; everyone else caches it and is told when it changes.

ML-infra tie
A feature store or a vector index sharded by entity id (user, item) is a textbook partitioned store. Hash-partition by entity id and online inference reads route to one partition — fast. But a "celebrity" entity (a viral video's embedding, a hyperactive user's features) is a hot key: 100 percent of its lookups hit one shard exactly as in §3, and the fix is the same — split the hot entity's key, or front it with a cache. A query that needs all entities in a region, or a vector search that has no entity-id filter, is scatter-gather across every shard. And when you add GPUs/replicas to grow the index, you want consistent hashing or fixed partitions so you re-index a slice, not the whole corpus.

Trade-offs

ChoiceBuysCosts
Range partitioningEfficient sorted range scans; only adjacent partitions touchedHot end if access concentrates (timestamps, sequential ids)
Hash partitioningEven spread of bytes and load; tames structured-key skewLoses sort order; range queries become scatter-gather
Compound key (hash + sort)Balanced across the hash part, local range scans within itQueries omitting the hash part still scatter-gather
Local secondary indexWrites stay on one partitionReads by secondary field scatter-gather every partition
Global secondary indexSecondary reads hit one partitionWrites fan out across partitions; index lags (async)
Hash mod NTrivially balanced while N is fixed~(N−1)/N of data moves whenever N changes — unusable
Fixed partitions / consistent hashingAdding a node moves only ~1/N of the dataMore partitions/virtual-nodes metadata to manage

Failure modes

  • Hot partition caps the cluster. One celebrity/viral key takes a large share; that single partition saturates while the rest idle. Symptom: throughput plateaus and tail latency spikes no matter how many nodes you add.
  • Timestamp/sequence hot end. Range-keying by "now" or an auto-increment id sends all writes to the newest partition. Symptom: one shard at 100 percent CPU, others cold.
  • Accidental scatter-gather. A common query omits the shard key, so every request fans out to all partitions. Symptom: latency tracks the slowest partition and grows with cluster size.
  • Rebalance storm. Naive hash-mod-N (or an aggressive auto-rebalancer) moves most of the data at once and starves live traffic. Symptom: a node addition tanks latency cluster-wide.
  • Stale routing map. A client caches an old partition-to-node map after a move and hits the wrong node. Symptom: errors or misroutes spiking right after a rebalance.
  • Global-index lag mistaken for truth. An async global secondary index hasn't caught up; a read returns rows that no longer match. Symptom: index disagrees with the primary just after writes.

Decision checklist

  • What is the dominant query? Put its filter in the shard key so it routes to one partition; everything else may scatter-gather.
  • Do you need range scans? If yes, range or compound key; if not, hash for balance.
  • Is any single key hot (celebrity, viral, hub entity)? Plan to split it (random suffix) or cache it — hashing alone won't save you.
  • Which secondary fields are queried, and is that side read-heavy (favor global) or write-heavy (favor local)?
  • Did the node count N leak into your placement function? If so, replace with fixed partitions or consistent hashing before you ever scale.
  • How do clients find a partition, and how are they told when the map changes? Rate-limit and observe rebalancing.
  • Each partition is a replica set — did you size n, w, r (lesson 09) per partition, not per cluster?

Checkpoint exercise

Try it
You operate an online feature store for a recommender, hash-partitioned by user_id across 8 partitions, serving 80,000 lookups per second. (a) Inference reads one user's features per request — show why this is a single-partition point lookup and state its throughput headroom under a uniform load. (b) One celebrity user suddenly draws 25 percent of all lookups; compute that partition's load, the cluster average, and the imbalance ratio, then propose a concrete fix and the read-path cost it adds. (c) The product team now wants "all users in region = EU" — explain why this is scatter-gather under your current key, and choose between a local and a global secondary index on region, justifying it from the read/write mix. (d) You add 2 nodes (8 → 10). Explain why hash(user_id) mod N would be catastrophic and which rebalancing scheme you'd use instead, with the rough fraction of data that moves.

Where this points next

This lesson built the mechanics of distribution — shard keys, hash vs range, hot-key splitting, secondary-index layouts, and rebalancing schemes. But mechanics are not operations. The moment you run these schemes for real — many tenants sharing one cluster, a routing tier in front of moving partitions, a rebalance executed on a live system at 3 a.m. — a new layer of questions appears: how do you keep one noisy tenant from starving the others, how does the router learn the map and stay current, and what is the actual runbook for splitting a hot shard without dropping writes? The next lesson, Multitenancy, Request Routing, and Sharding Operations (lesson 12), is that operations layer built directly on these mechanics: it takes the placement and rebalancing primitives you just learned and turns them into tenant isolation, routing topologies, and the rebalance runbooks that keep a partitioned cluster healthy while it is serving traffic.

Where is truth?
Under partitioning, truth is located, not centralized — and two different things hold authority. System of record: for any key, the partition that owns it (precisely, the leader/quorum of that partition's replica set); for "which partition owns which key," the authoritative partition-to-node map held by the coordination service (ZooKeeper/etcd) or the routing tier. Copies / derived views: the replicas of each partition (the redundancy axis from Part 4), plus secondary indexes — local indexes are per-partition derived copies, a global index is a separately-partitioned derived view. Freshness budget: the routing map's cache TTL (how stale a client's view of partition placement may be after a rebalance) and, for async global indexes, the index-lag budget behind the primary. Owner: the partition owner for the data; the cluster controller/coordination service for the placement map. Deletion path: a delete routes to the owning partition by shard key and must also purge every secondary index entry — a local index deletes alongside the row, a global index deletes asynchronously and can briefly return stale matches. Reconciliation / repair path: rebalancing moves whole partitions (fixed-count or consistent-hashing) to restore balance; hot-key splitting redistributes a celebrity key; clients refresh a stale routing map on a misroute. Evidence it is correct: the imbalance ratio (max-over-average) staying near 1, no partition pinned at 100 percent CPU while peers idle, every key resolving to exactly one owning partition, and the global index's lag staying inside its budget.
Takeaway
Partitioning splits the data itself across machines (orthogonal to replication, which copies it), so capacity and load scale with node count — but only if load actually spreads. The shard key is a performance contract: range partitioning keeps keys sorted (cheap range scans, but a hot end when access concentrates), while hash partitioning spreads load evenly (no range scans, range queries become scatter-gather), and a query that omits the shard key must fan out to every partition. Hashing tames structured-key skew but not a single hot key — a celebrity's traffic all shares one key and lands on one partition (worked: a 30 percent hot key on 10 partitions gives a 3.7× imbalance), fixed only by splitting the key or caching it. Secondary indexes mirror the trade (local = local writes / scatter-gather reads; global = scatter writes / targeted reads). And because hash(key) mod N moves nearly all data when N changes, real systems use a fixed partition count, consistent hashing, or dynamic splitting so adding a node moves only about 1 / N of the data, with a routing layer telling clients where each partition now lives.

Interview prompts