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, customer or product gets millions of times more traffic than the rest. Think of a pop star whose followers outnumber everyone else's, a product in a flash sale, or one huge business customer. Hashing does not help, because it is still just one key. Cures:
  • Split the key. Add a random ending, from user123_0 to user123_9, so the hot key becomes ten keys. Reads then ask all ten and combine the results. This works very well for counters and append-only data, and badly for anything that needs one ordered view.
  • Put a cache in front. A key that is read constantly belongs in a cache, not a shard (7.6).
  • Give it its own partition. The giant gets its own machines, with a special routing rule.
  • Rethink the model. For a celebrity's followers, instead of pushing each post to every follower when it is written, followers pull it when they read. The news-feed design in Chapter 11.8 does exactly 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? There are three ways to route:

• A routing layer. A proxy in the middle knows the map. This is the classic approach.
• Routing in the client. The client library holds the map. That saves a hop, but every client must be updated when the map changes.
• Any node. Send the request anywhere, and the node that receives it passes it on. Cassandra works this way.

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).

Rules for rebalancing:

• Move partitions slowly, with a speed limit. A rebalance that fills the network is the outage.
• Keep both copies until the new one has been checked.
• Never rebalance automatically because a node is briefly missing. A node down for a 5-minute restart should not cause terabytes of data to move. That is 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.
• The costs:
  • Queries that ask every partition. The slowest partition sets your response time.
  • Transactions across partitions need 2PC or sagas. Design to avoid them.
  • Secondary indexes. A local index means asking every partition on reads. A global index means writes spread across machines.
  • Joins turn into copying data into each place that needs it.
  • Resharding is a risky migration done while the system is live.
• 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)

Give it its own partition or machines. Send the known giants to their own capacity, using a table of special routing rules. You accept some special cases in operations, and in return a problem with one giant cannot spread. Combine this with (1). (4) Change the model. This is the real fix for feeds. Stop pushing the celebrity's post into 50 million followers' inboxes when it is written. Instead, followers pull celebrity posts when they read, and merge them into the rest of their feed. That is the mixed design of Chapter 11.8. The hot spot disappears, because the work moves to the readers, who are already spread across many machines. (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.

Partition by tenant_id, for three reasons:

• Almost all access is per customer. Every everyday query filters by tenant, and transactions naturally touch one tenant, so they stay inside one partition. That avoids the cost from section 5 by design.
• It keeps customers apart, so one noisy customer does not slow the others.
• It matches the security boundary. A query with a missing filter cannot reach across shards.

Not order_id, because one customer's data would be scattered everywhere. Not time, because of 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, with no downtime:

1. Add the routing layer to the app while everything still points at the one database. Nothing changes for users, and the routing logic gets tried on real traffic.
2. Set up shards 1 to N.
3. Write to both for one customer at a time. Writes go to the old and the new database, and reads still come from the old one. A background job copies the existing data across, and another keeps checking that row counts and checksums match.
4. Switch that customer's reads to the new shard, using a flag per customer. If anything goes wrong, only one customer is affected, and you can switch back at once.
5. Stop writing to both. After a waiting period, delete that customer's data from the old database.
6. Repeat. Either start with the biggest customers, because they cause the pain and prove the process, or with the smallest, because they are lower risk to learn on. Say which you chose, and why.

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 quietly worked across all customers. For example:

• rules that something must be unique everywhere, such as order numbers and invoice sequences;
• foreign keys that point from one customer's data to another's;
• admin queries that scanned everything;
• SELECT MAX(id).

Each needs an explicit replacement: per-customer sequences or IDs that are unique everywhere (Chapter 11.5), and admin tools that read from the warehouse. Set aside real time to find these. You find them first by searching the code, and then by surprises in production. They are the reason sharding projects run late.

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 two-part key from section 1, and each part has a job:

• Hashing the conversation ID spreads the load evenly across shards. There are many conversations, and they are roughly similar in size, unlike users.
• Sorting inside the partition makes the most common query, "the last 50 messages in this conversation, then scroll back", a single range read in one partition. That is the cheapest read there is.

It also gives you ordering for free. One partition means one writer per conversation, so message numbers are exact, and no machinery is needed to agree on order across machines (10.3).

Big groups are fine. A group of 10,000 members is still one conversation with one stream of messages. It is the members' reads that spread 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 keep a second index, partitioned by user (partition by user_id, sorted by most recent activity), and update it whenever a message is sent. It is a read model kept up to date from the messages (10.8.4), with a clear owner and a clear way to rebuild it (10.4).

Notice what it costs. Sending one message to a group of 10,000 writes 10,000 index rows. That spreading out must happen in the background, in batches, and be safe to repeat. For very large groups, switch to working out the list when it is read, from the group's members. That is the celebrity-key trade-off from section 3 again. Here it shows up as a group size where you switch, which 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.