10.6 — Partitioning: Splitting Data So Writes Scale

Replication (10.5) gives every node the whole dataset — which is why it multiplies read capacity and does nothing for writes. Partitioning (sharding) does the opposite: it splits the data so each node holds a subset, so writes, storage, and working-set size divide across the fleet. The entire craft is in one decision — the partition key — which determines whether your system scales linearly or develops a hot spot that no amount of hardware fixes. This page covers the strategies, consistent hashing (visualized), hot spots and their cures, rebalancing, and the operations partitioning makes expensive.

1. Two strategies, and what each costs

Range partitioning — contiguous key ranges per partition (A–F here, G–M there; or by time: this month, last month). Advantage: range scans are efficient — "all orders in March," "keys between X and Y" hit one or few partitions. Cost: hot spots by construction when keys are sequentially generated — partition by timestamp and every write goes to the newest partition while the others idle; partition by auto-increment id and you've built a single-writer system with extra machines. (HBase and Bigtable use ranges; both come with hot-spot folklore.)

Hash partitioning — hash(key) mod N (or better, consistent hashing). Advantage: uniform distribution, so writes spread evenly regardless of key shape. Cost: range scans are gone — adjacent keys land on different nodes, so "all orders in March" becomes a scatter-gather across every partition. Also: the naive mod N form is a rebalancing disaster (section 3).

The practical hybrid most systems use: a compound key — hash a high-cardinality prefix for distribution and range-order the rest within it. Cassandra's model is exactly this (partition key + clustering columns): (user_id) → hash for placement, (timestamp) → sorted within the partition, giving even spread and efficient per-user time-range queries. When you can shape the key this way, do.

2. Consistent hashing

The problem with hash(key) mod N: change N (add or remove one node) and almost every key remaps — a full data reshuffle and a cold cache fleet-wide. Consistent hashing fixes it by mapping both keys and nodes onto a ring: a key belongs to the first node clockwise from its position, so adding a node steals keys only from its immediate neighbor — roughly 1/N of the data moves, not all of it.

Two refinements are essential in practice. Virtual nodes: each physical node claims many points on the ring (100–256 typically), which smooths distribution (raw random placement of few points is lumpy) and spreads the load of a departing node across all remaining nodes rather than dumping it on one neighbor. Heterogeneous weighting: a machine with twice the capacity claims twice the virtual nodes.

Figure 1 — Consistent hashing. Keys and (virtual) nodes share a ring; a key belongs to the first node clockwise. Joining or leaving moves only the neighboring slice — roughly 1/N of the data — instead of the near-total remap that mod N forces.

3. Hot spots: the failure that hardware can't fix

A hot spot is a partition receiving disproportionate traffic — and the defining property is that adding nodes doesn't help, because the load is concentrated on one key or range. The recurring causes and their cures: ^{⚑Consistent hashing and hot partitions. [EQ-1095b]}

• Sequential keys (timestamps, auto-increment ids) with range partitioning → all writes to the newest partition. Cure: hash the key, or prefix it with a hashed/random component (shard_id:timestamp).
• Celebrity keys — one user, tenant, or product with millions of times the traffic (the Bieber problem: one account's followers dwarf the rest; a flash-sale SKU; a single enterprise tenant). Hashing does not help — the key is one key. Cures: (a) key splitting — append a random suffix (user123_0 … user123_9) so the hot key becomes ten keys, with reads fanning out and merging (works beautifully for append/counter workloads, badly for anything needing a single ordered view); (b) caching in front — a hot read key belongs in a cache tier, not a shard ([7.6]); (c) dedicated partition — give the whale its own node(s) and route by exception; (d) rethink the model — for celebrity fan-out, push-on-write becomes pull-on-read (Chapter 11.8's news-feed hybrid is precisely this).
• Low-cardinality keys — partitioning by country when 80% of users are in one country, or by status when 95% of rows are active. Cure: choose a higher-cardinality key or a composite.

The diagnostic to build early: per-partition metrics (requests, bytes, latency), not just aggregates. A fleet at 30% average CPU with one shard at 100% looks healthy on every dashboard that averages — and that shard is your outage.

4. Rebalancing and routing

How many partitions? The robust answer is fixed, over-provisioned partitions (e.g. 1024) mapped onto however many nodes you have; scaling moves whole partitions between nodes without splitting or re-hashing keys. This is Kafka's model (partition count fixed at topic creation — and raising it is disruptive because it changes key→partition mapping) and Elasticsearch's (shards fixed per index). Dynamic splitting (HBase, DynamoDB) splits a partition when it grows past a threshold — more automatic, more operationally surprising.

Who knows where a key lives? Three routing models: a routing tier (a proxy that knows the map — the classic middleware); client-side routing (the client library holds the map — one fewer hop, but every client must be updated on changes); or any-node routing (send anywhere, the receiving node forwards — Cassandra's coordinator model). The map itself needs a home. That is usually a small, strongly consistent coordination store whose only job is to hold facts every node must agree on — etcd and ZooKeeper are the two you will meet — or gossip between the nodes themselves, or a control plane (10.7.2).

Rebalancing discipline: move partitions gradually (throttled — a rebalance that saturates the network is the outage), keep both copies until the new one verifies, and never rebalance automatically on a node's temporary absence (a node down for a 5-minute restart should not trigger terabytes of movement; that's the failover-storm lesson again — 10.5).

5. What partitioning makes expensive

The honest costs, which are why partitioning is a decision and not a default:

• Cross-partition queries become scatter-gather: query every partition, merge results — latency is the slowest partition's (tail amplification: with 100 partitions, a p99 event on any one of them becomes your median), and cost rises with fan-out.
• Cross-partition transactions need distributed commit (2PC — 10.7.2) or saga-style compensation (10.8.4). Most systems design to avoid them: choose a key such that transactions are naturally single-partition (all of a customer's data in one place).
• Secondary indexes get hard: a local index (per partition) requires scatter-gather to query; a global index (partitioned by the indexed field) is fast to read but must be updated across partitions on write — a distributed write with its own consistency question.
• Joins across partitions are expensive enough that denormalization becomes the norm ([7.5]/[7.8]).
• Resharding is one of the most dangerous operations in production: a live key-space migration with dual writes, backfill, verification, and cutover (10.11).

6. The expert lens

The partition key is the highest-leverage decision in a data-intensive design, and it is nearly irreversible. It determines write distribution, which queries are cheap, which transactions are local, and where hot spots can appear. Choosing it well means simulating the actual access patterns first: the top ten queries by volume, the write distribution by key, the largest tenant's share. Choosing it badly means a resharding project, which is why interview answers should always state the key and the access patterns that justify it — "partition by customer_id, because every transaction and 95% of reads are customer-scoped, and no customer exceeds 2% of volume" is a complete answer; "we'll shard the database" is not.

Locality is the goal; uniformity is the constraint. You want related data together (single-partition transactions, cheap queries) and load spread evenly (no hot spots), and these pull in opposite directions — perfect locality is one partition, perfect uniformity is random placement. Every good key is a negotiated point between them, which is exactly why compound keys (hash for spread, range within) are so common.

Partitioning and replication are independent and both are required at scale. Each partition is itself replicated (leader plus followers, or a quorum group), so a real cluster is an N×M grid: partitions divide the write load, replicas of each partition provide durability and read capacity. Systems people often conflate them — "we added replicas but writes are still slow" (10.5) and "we sharded but a node failure loses data" are the two symmetric confusions, and holding both axes clearly is what lets you diagnose either.

Next: 10.7.1 — with data copied and split, the question that decides how it behaves: what consistency actually means, CAP and PACELC honestly, and the model spectrum from linearizable to eventual.

Recall

• Partitioning/sharding splits data so writes, storage, and working set divide (replication cannot do this). Range partitioning: cheap range scans, hot spots with sequential keys. Hash: uniform spread, no range scans. Practical hybrid: compound key — hash prefix for placement, range-ordered suffix within (Cassandra's partition key + clustering columns).
• Consistent hashing: keys and nodes on a ring, key → first node clockwise ⇒ join/leave moves ~1/N of data instead of mod N's near-total remap. Essential refinements: virtual nodes (smooth spread, distribute a departing node's load) and capacity weighting.
• Hot spots are where adding nodes doesn't help: sequential keys (hash or prefix them), celebrity keys (split with random suffixes, cache in front, dedicate a partition, or change the model — feed fan-out becomes pull), low-cardinality keys. Build per-partition metrics — averages hide the one shard at 100%.
• Rebalancing: prefer fixed over-provisioned partitions moved whole (Kafka/ES) over dynamic splitting; routing via routing tier / client-side map / any-node forwarding, with the map in etcd/ZooKeeper/gossip; rebalance throttled, verified, and never on transient absence.
• Costs: scatter-gather queries (tail amplification — the slowest partition sets your latency), cross-partition transactions (2PC or sagas — design to avoid), secondary indexes (local = scatter reads; global = distributed writes), joins → denormalization, and resharding as a dangerous live migration.
• Lens: the partition key is the highest-leverage, near-irreversible decision — justify it with real access patterns; locality vs uniformity is the trade every key negotiates; partitioning and replication are independent and both required (an N×M grid).

Self-test: Why does timestamp range-partitioning hot-spot, and what fixes it? What exactly does consistent hashing improve over mod N, and what do virtual nodes add? Name three cures for a celebrity key. Why is a fixed large partition count operationally safer? What becomes expensive after partitioning — five items?

Quiz Bank

FoundationalRange vs hash partitioning: the trade, and the hybrid that usually wins.

Range partitioning assigns contiguous key ranges to partitions, so range scans are efficient — "all events in March," "keys A–F" touch one or few partitions — which is why time-series and ordered workloads reach for it. Its structural weakness is hot spotting on sequential keys: partition by timestamp and every write lands on the newest partition while the rest idle (the same for auto-increment ids), producing a single-writer system with extra machines.

Hash partitioning distributes hash(key) across partitions, giving uniform load regardless of key shape — but destroys ordering, so range queries become scatter-gather across every partition, and the naive mod N form makes rebalancing catastrophic (section 2).

The hybrid: a compound key that hashes a high-cardinality component for placement and keeps a sorted component within the partition — Cassandra's partition key plus clustering columns, DynamoDB's partition key plus sort key. (user_id, timestamp) spreads users evenly and answers "this user's events in March" from one partition. When your access patterns are entity-scoped with a time or sequence dimension — which is most of them — this is the shape to reach for first.

FoundationalWhat problem does consistent hashing solve, and why are virtual nodes necessary?

The problem: with hash(key) mod N, changing N remaps almost every key — adding one node to a 10-node cluster moves roughly 90% of the data, saturating the network, cooling every cache, and making scaling an outage. Consistent hashing places both keys and nodes on a hash ring; a key belongs to the first node clockwise. Adding or removing a node therefore affects only the arc between it and its predecessor: about 1/N of keys move, and the rest stay put.

Virtual nodes fix two residual problems: (1) lumpy distribution — with only one ring point per node, random placement gives some nodes much larger arcs than others (variance is high with few samples); assigning 100–256 virtual points per physical node averages this out to near-uniform. (2)

Failure concentration — without virtual nodes, a departing node dumps its entire range onto exactly one successor, which then carries double load precisely when the cluster is already degraded; with virtual nodes, its many small arcs distribute across all remaining nodes. Virtual nodes also enable heterogeneous capacity: a machine with twice the resources takes twice the ring points. Used by Cassandra, DynamoDB, Riak, and most consistent-hash load balancers (9.7.6's sticky routing).

AppliedA social platform partitions by user_id. One celebrity account with 50M followers makes its shard fall over. Walk the cures.

Hashing doesn't help — the load is one key, so every request maps to one partition by definition. Cures, in increasing order of change: (1) Cache in front — most celebrity traffic is reads of the same data; a cache tier absorbs it and the shard sees a trickle ([7.6]); this is the fastest mitigation and often sufficient. (2)

Key splitting for the write-heavy parts — append a bucket suffix (celeb_0 … celeb_31) so writes (new posts' fan-out records, counters, likes) distribute across 32 partitions; reads fan out and merge. Excellent for append-only and counter workloads, poor where a single ordered view is required — which is the trade to state. (3)

Dedicated partition/nodes — route known whales to their own capacity by exception (a routing override table), accepting operational special-casing in exchange for isolating the blast radius; combine with (1). (4) Change the model — the real fix for feeds: stop pushing the celebrity's post into 50M follower inboxes on write (fan-out-on-write), and instead let followers pull celebrity content at read time, merging it with their pushed feed (the hybrid design of Chapter 11.8) — the hot spot disappears because the work moves to the readers, who are already distributed. (5)

Rate-limit and degrade the celebrity's non-essential derived work (analytics rollups, notification fan-out) so it can't compete with serving. Whatever the mix, add per-key/per-partition metrics first — the shard was at 100% while the fleet averaged fine, and you cannot manage what averages hide.

InterviewWhat becomes hard after you partition? Enumerate the costs.

(1) Cross-partition queries turn into scatter-gather: fan out to every partition, merge, and wait for the slowest — so your latency becomes the tail of N partitions (with 100 shards, a p99 hiccup on any single one lands on most requests — tail amplification), and cost scales with fan-out.

(2) Cross-partition transactions require distributed commit (2PC — blocking, and a coordinator failure leaves locks held — 10.7.2) or saga-style compensation (10.8.4); the design response is to choose a key that makes transactions single-partition by construction (all of a customer's data co-located).

(3) Secondary indexes: a local index (each partition indexes its own rows) needs scatter-gather on query; a global index (partitioned by the indexed value) reads fast but every write must update a remote partition — a distributed write with its own consistency and failure story.

(4) Joins across partitions are expensive enough that denormalization becomes standard practice ([7.5]). (5) Resharding — changing the key or partition count on live data is among the riskiest production operations: dual writes, backfill, verification, cutover, and rollback planning (10.11).

(6) Operational complexity — per-partition monitoring, uneven growth, rebalancing throttles, and the fact that "the database is fine" is now a per-shard statement. The summary judgment: partition when write volume, data size, or working set genuinely exceeds one machine — and pick the key so that the common transactions and queries stay inside a single partition.

StaffYou must shard a 4 TB single-Postgres order system serving a B2B SaaS: heavy per-tenant queries, monthly reporting across all tenants, a few tenants 100× larger than the median, and no acceptable downtime. Choose the key, plan the migration, and name the two things that will hurt.

Key: tenant_id — because access is overwhelmingly tenant-scoped (every operational query filters by tenant, and transactions are naturally single-tenant, keeping them single-partition — the section 5 cost avoided by design), it gives natural isolation for noisy-neighbor control, and it aligns with the security boundary (a mis-scoped query can't cross shards). Not order_id (destroys locality), not time (hot spots, section 3).

Handling the 100× tenants: pure hash(tenant_id) would place whales randomly, so shards holding one become imbalanced — instead map tenant → shard through a lookup table (a directory-based assignment, not a pure hash), letting you place whales on dedicated shards and pack small tenants together, rebalancing individual tenants without touching anyone else. The lookup table is a control-plane concern (etcd/Postgres with heavy caching), and it also makes tenant migration a first-class operation rather than a re-hash.

Migration plan (no downtime): (1) introduce the routing layer in the app while everything still points at the single database (behavior-preserving, verifies routing logic under real traffic); (2) stand up shard 1..N; (3) dual-write for one tenant at a time — writes go to old and new, reads still old — with a background backfill and a continuous verification job comparing row counts and checksums; (4) flip that tenant's reads to the new shard behind a per-tenant flag (blast radius = one tenant, instantly revertible); (5) stop dual-writing, remove the tenant's data from the old database after a retention window; (6) repeat, whales first (they cause the pain and prove the process) or smallest first (lower risk to learn on) — state the choice and its rationale.

The two things that will hurt: (a) Monthly cross-tenant reporting — now a scatter-gather over N shards with tail amplification and no cross-shard joins. Do not solve this by keeping reporting on the operational shards: replicate to an analytical store (columnar warehouse via CDC/ETL — [7.8]) where cross-tenant queries belong; this is the moment the read model and the write model correctly diverge (10.8.4's CQRS instinct). (b)

Anything that was implicitly cross-tenant — global uniqueness constraints (order numbers, invoice sequences), foreign keys spanning tenants, admin queries that scanned everything, and SELECT MAX(id) patterns; each needs an explicit replacement (per-tenant sequences or globally-unique ids — Chapter 11.5's ID generation, admin tooling pointed at the warehouse). Budget real time for discovering these — they are found by grep and by production surprises, in that order, and they are the reason sharding projects overrun.

Flashcards

FlashRange vs hash

Range: cheap scans, hot-spots on sequential keys. Hash: uniform, no scans. Hybrid: hash prefix + sorted suffix (Cassandra partition + clustering key).

FlashConsistent hashing

Ring: key → first node clockwise ⇒ join/leave moves ~1/N (vs mod N's total remap). Virtual nodes = smooth spread + distributed failure load + capacity weighting.

FlashHot spot cures

Sequential keys → hash/prefix. Celebrity key → cache, split with suffixes, dedicate a shard, or change the model (push→pull). Low cardinality → higher-cardinality key.

FlashPartition count

Prefer fixed, over-provisioned partitions moved whole (Kafka/ES) over dynamic splits. Routing map in etcd/gossip; rebalance throttled, verified, never on transient absence.

FlashPost-partition costs

Scatter-gather (tail amplification) · cross-partition transactions (2PC/saga) · secondary indexes (local vs global) · joins → denormalization · resharding.

FlashTwo axes

Replication = availability + read scale (every node has everything). Partitioning = write/storage scale (each node has a subset). Real clusters are N×M — both.

Scenario Drill

DrillDesign the partitioning scheme for a chat platform: 1-1 and group conversations, messages must be ordered within a conversation, users read their own conversation list constantly, groups can have 10k members, and search across all of a user's messages is a feature. Specify keys per access pattern, and resolve the conflict the last requirement creates.

Message storage — partition by conversation_id, cluster by (timestamp, message_id). This is the compound-key hybrid (section 1): hashing the conversation spreads load evenly across shards (conversations are numerous and roughly comparable in size, unlike users), while sorting within the partition makes the dominant query — "the last 50 messages in this conversation, then scroll back" — a single-partition range scan, which is the cheapest possible read. It also gives ordering for free: a single partition means a single writer per conversation, so sequence numbers are exact and no cross-node ordering machinery is needed (10.3's per-key ordering, purchased by the key choice). Large groups are fine — a 10k-member group is still one conversation with one message stream, and it's members' reads that fan out, not the writes.

Conversation list per user — a separate, differently-partitioned view. "Which conversations do I have, with unread counts and last message preview" is a per-user query, which the message partitioning cannot serve without scanning everything. So maintain a derived, user-partitioned index (partition by user_id, sorted by last-activity) updated when messages are sent — a materialized read model (10.8.4) whose ownership and rebuild path are explicit (10.4). Note what this costs: sending one message to a 10k-member group writes 10k index rows — a fan-out that must be asynchronous, batched, and idempotent, and that for very large groups switches to pull-on-read (compute the list from group membership at read time) — the celebrity-key trade-off (section 3), appearing here as the group-size threshold you tune.

The conflict — search across all of a user's messages — genuinely fights both keys: it's neither conversation-scoped nor cheaply derivable from a user-partitioned list, and running it as a scatter-gather over every conversation partition is exactly the tail-amplified query section 5 warns about.

Resolution: search is not a partition problem, it's a different system. Feed messages into a search index ([7.7]) partitioned by user_id (so a user's search hits one or few shards), storing only what search needs (tokens, ids, timestamps, conversation ids — not full message bodies if privacy/retention argue against it). This is the standard and correct move: the operational store is partitioned for writes and conversation reads; the search store is partitioned for the search access pattern; both are fed from the same source of truth with an explicit, rebuildable pipeline.

The design-doc line: one access pattern, one partitioning scheme — messages by conversation for ordering and writes, conversation lists by user for the inbox, search by user in a dedicated index — with the fan-out cost of the derived views named, bounded, and switchable to pull-on-read for the largest groups.