Deep Read · DDIA · Chapter 6
Designing Data-Intensive Applications · Ch 6 · Martin Kleppmann · 2017
The last chapter was about keeping several copies of the same data (replication). This chapter tackles a different wall: the data itself is too big—too big for any single machine's disk, and too busy for its CPU. A social network's billions of messages, an online store's hundreds of millions of products—no one server can hold or serve all of it. The fix is blunt: chop the giant pile into many small chunks and give each chunk to one machine. That chopping-and-spreading act is called partitioning (also known as sharding).
Imagine a huge library, far too many books for one shelf. You have two ways to split them. One: by the first letter of the title—A through F on shelf 1, G through M on shelf 2, and so on. Two: give every book a scrambled "code number"—the same title always gets the same code—and drop it on the shelf its code points to. The first makes "find all the D books" easy (they're together); the second makes every shelf equally full so none gets crowded. Databases split data with exactly these two moves.
Not splitting is a dead end: one machine can't store that much, and can't keep up with that many requests. But the moment you split, a new problem appears—an uneven split causes traffic jams. If all the popular requests happen to land on the same shelf (a celebrity posts, and tens of millions arrive at once), that one machine gets crushed while the others sit idle. This "one cell overloaded, the rest empty" situation is the enemy the whole chapter keeps fighting.
1. How do you split it evenly? Splitting by first letter is simple and lets you grab a range in order—but it invites hot spots (everyone wants the newest thing). Splitting by scrambled code is the most even—but you lose the ability to "grab a range in order" (neighbors got flung to opposite ends). Each has its sweet spot and its sting.
2. When you add a new shelf, how do you move the books? When the library expands, you can't tear down and re-sort everything—that means closing for days. The smart move is to shift only a small fraction of books from each old shelf to the new one; move as little as possible.
3. To find one book, how do you know which shelf it's on? You need a "front desk" or a "directory" that tells you which machine owns the data you want. And when shelves shift, that directory has to stay current.
No—but they're a golden pairing. Replication is "the same content, copied onto several machines" (survive failure, stay close, spread reads). Partitioning is "chop the whole big dataset into different chunks, each machine owns one chunk" (so it fits and keeps up). Real systems almost always do both at once: chop into many chunks, then keep three copies of each chunk on different machines—so it fits and it can survive failures.
Partitioning = chop a dataset too big for one machine into many small chunks, each owned by one machine, so you can scale. Two ways to chop: in order (easy to grab ranges, but jam-prone) or by scrambled code (most even, but no range grabs). The real difficulty is keeping any one chunk from getting crushed (hot spots), moving little data when you add machines, and always knowing where each record lives. It's usually used together with last chapter's replication.
Want the actual mechanisms, secondary indexes, rebalancing strategies, and diagrams? → Switch to the deep read
Partitioning (a.k.a. sharding) means splitting a large dataset into non-overlapping chunks—partitions—each living on one node, so that storage and query load spread across many machines. It is the core mechanism for scaling data. The whole chapter's tension sits in four questions: how to split evenly (key-range vs hash), how to stop any one partition from being crushed (hot spots / skew), what to do about secondary indexes (local vs global), and how to re-spread partitions—moving as little data as possible—when nodes are added or removed (rebalancing).
This is the second chapter of Part II, "Distributed Data," right after Chapter 5 on replication. The two are complementary ways to scale horizontally, and they usually stack: replication puts whole copies of the data in several places; partitioning slices the dataset and puts each slice in one place. Real systems almost always "partition first, then replicate each partition a few times" (see Figure 3). Partitioning also opens the next layer of difficulty: an operation that spans several partitions needs cross-partition consistency and coordination—which is exactly what Chapter 7 (transactions) and Chapter 9 (consistency and consensus) pick up. Think of this chapter as swapping Chapter 5's "many replicas" problem for a "many shards" one.
Replication answers "what if a node dies" and "what if there are too many reads," but it does nothing about the data being too big—every replica still stores the full dataset. Picture a social app: a billion users, tens of TB of data, write peaks of hundreds of thousands per second. At that scale two walls hit at once: one disk can't hold tens of TB, and one machine's CPU / memory / disk I/O can't absorb hundreds of thousands of writes per second. Adding replicas doesn't help—each replica must store everything and swallow every write.
The way out is to split: each node stores only 1/N of the data and handles only the requests that land on its chunk, so adding machines can scale you out roughly linearly. But splitting introduces problems the single-machine era never had, and this chapter solves each: 1. By what rule do you split, so both data and load spread evenly with no hot spots? 2. How do secondary indexes (non-primary-key queries) work on split data? 3. When the cluster grows or shrinks, how do you re-spread partitions—moving as little data as possible, without downtime? Split badly, and scaling out buys you not performance but one crushed partition dragging down the whole system.
Partitioning's only enemy is skew: data or requests piling disproportionately on some partitions, which then become hot spots—the very thing scaling out was meant to avoid. The dumbest scheme, "assign at random," spreads perfectly evenly, but then you don't know where anything is and must ask every partition on every read—unusable. So a real scheme has to balance "spreads evenly" against "findable," which gives us the two mainstream ways to split below.
What it is: divide keys into contiguous ranges, each partition owning one range—like a printed encyclopedia split by letter (A–C in one volume, D–F in another). Boundaries can be chosen by hand or picked automatically by the database (to keep each range's data volume roughly equal). Within a partition keys are stored in sorted order (with LSM-trees / SSTables), so range scans are very efficient: "all readings from one sensor on one day" is just a contiguous sweep of a single partition. Used by: Bigtable, HBase, RethinkDB, early MongoDB.
Its fatal weakness is hot spots: if the key has a rising trend—the classic case being a timestamp key—then every new write slams into the "today" partition while yesterday's sits idle and today's gets crushed. DDIA's fix is to prefix the key to spread it out: change the key from "timestamp" to "sensor-name + timestamp," so writes fan out by sensor name first; the cost is that querying "all sensors over a time window" now means scanning a stretch per sensor and merging.
What it is: run the key through a hash function to get an evenly spread number, then partition by ranges of that hash value. A good hash turns "skewed keys" into "uniform hash values"—even a mass of very similar keys lands evenly across partitions once hashed. Used by: Cassandra, MongoDB (hashed sharding), Voldemort.
The cost is losing range queries: neighboring keys (like consecutive timestamps) get scattered across partitions, so "grab a contiguous range" must ask every partition. Cassandra's compromise is a classic: a compound primary key—the first column (the partition key) is hashed to pick the partition, while the remaining columns are stored in sorted order within that partition and support range scans. So "all of one user's posts sorted by time" ranges efficiently inside a single partition, but across users it can't. This is the general "hash by A, then keep sorted by B within each A" pattern.
Aside: the often-cited consistent hashing was introduced by Karger et al. in 1997 for CDN / web caching, arranging hash values on a ring to minimize data movement when nodes join or leave. DDIA specifically warns that what databases call "hash partitioning" is usually not consistent hashing in that paper's strict sense—the term is used loosely, so don't get tangled up in it.
Hashing spreads out different keys; but reality often has one key hammered relentlessly—a celebrity account posts, tens of millions read and write the same record. That record's key hashes to just one partition, so that machine gets crushed anyway and hashing does nothing. DDIA is frank: most systems today can't automatically compensate for such a "hot key"—the application must step in. The common move is salting the hot key: append a small random number (say 0–99), artificially splitting one hot record into 100 keys scattered across 100 partitions to share the writes. The cost is that reads must fetch all 100 and merge them, and you have to remember which keys were split—a workaround with real bookkeeping.
So far we've split by the primary key. But real queries often ask "all red cars" or "all posts by Zhang San"—that's the job of a secondary index. The snag: a secondary index doesn't cleanly follow the partitions the way the primary key does, because "red cars" may be scattered across every partition. DDIA offers two approaches—the pair most likely to be tested and most easily confused:
Approach A · Local (document-partitioned) index: each partition indexes only its own data, minding its own business. Writes are simple—changing a record touches only the one partition it lives on (data and its index sit together). But reads by secondary index are expensive: because "red cars" are spread across all partitions, you must send the query to every partition and merge the results—this is scatter / gather, whose read latency is held hostage by the slowest partition, so tail latency is easily amplified. Used by: MongoDB, Cassandra, Elasticsearch, SolrCloud, Riak, VoltDB.
Approach B · Global (term-partitioned) index: partition the index itself, but not by which partition a document lives in—by the indexed value (the term). For example, "all index entries for color=red" sit together in index partition 1, "color=blue" in index partition 2. Reads are fast: finding red cars asks only the one index partition "holding the red term," not all of them. But writes get complex: changing one record may touch several index partitions at once (its color, price, and brand terms each live in different partitions), so a single write becomes a distributed write across partitions—which is why global-index updates in practice are often asynchronous, meaning the index may lag briefly right after a write. Used by: DynamoDB's global secondary indexes (GSI).
Every design point in this chapter is a trade-off, and they are independent and freely mixed (split method × index method × rebalancing strategy). The three tables below lay out "which to pick when."
Table 1 · Key-range vs hash partitioning
| By key range | By hash | |
|---|---|---|
| Evenness | Depends on key distribution, skew-prone | Naturally even (a good hash flattens skewed keys) |
| Range queries | Efficient (sorted within partition, sweep a stretch) | Poor (neighbors scattered, must ask all) |
| Write hot spots | Rising keys (timestamps) all slam the last partition | No hot spot across keys; but a single hot key still crushes one partition |
| Typical use | Time series / range scans (sensors, log windows) | Point lookups, want evenness (user profiles, KV) |
| Systems | HBase, Bigtable, RethinkDB | Cassandra, MongoDB (hashed), DynamoDB |
Table 2 · Local (document-partitioned) vs global (term-partitioned) index
| Local index | Global index | |
|---|---|---|
| Index split by | Follows the document's partition | By the indexed value (term), split separately |
| Write | Simple: touches one partition | Complex: may span partitions, often async |
| Read by 2nd index | Expensive: scatter/gather across all | Fast: only the partition holding the term |
| Consistency | In sync with primary data | When async, may briefly read a stale index |
| Systems | MongoDB, Elasticsearch, Cassandra, Riak | DynamoDB global secondary index (GSI) |
Table 3 · Four rebalancing strategies (how to re-spread partitions when nodes change)
| Strategy | How it works | Cost / pitfall |
|---|---|---|
| hash mod N | partition = hash(key) % node count | Never use it: change the node count and N changes, so almost every key must move |
| Fixed number of partitions | Create far more partitions than nodes up front (e.g. 1000 partitions for 10 nodes); each node owns a batch; adding a node steals a few whole partitions from each existing node | Only whole partitions move, no re-hashing; but the partition count is fixed at setup and hard to change—too high wastes overhead, too low caps how far you can scale |
| Dynamic partitioning | Partition count adapts to data volume: a partition over a threshold splits, a shrunk one merges (like a B-tree) | Adapts to volume; but an empty DB starts with one partition, so early on all load hits one node (pre-splitting helps) |
| Proportional to nodes | A fixed number of partitions per node; a new node randomly splits existing partitions and grabs half | Partition count grows with nodes; random splits may not divide evenly |
Three practical rules of thumb: 1. Don't use hash mod N—it's the textbook anti-pattern for rebalancing, moving everything the moment you add a node. 2. Have "more partitions than nodes" but not too many: in the fixed-count scheme a partition is the smallest unit of movement, so too few caps scaling while too many piles up per-partition metadata / overhead. 3. Keep a human gate on rebalancing—fully automatic rebalancing plus automatic failure detection is dangerous: a node that's merely slow gets misjudged as "dead," triggering rebalancing → moving data adds load → more nodes look dead → a cascading avalanche. A "click to confirm" step blocks most such incidents.
One question remains after splitting: when a client wants to read a key, how does it know which node to connect to? (And partitions move around under rebalancing, so this mapping changes.) DDIA lists three approaches: 1. Let the client contact any node, which either serves the request or forwards it to the right node; 2. Add a routing tier that all requests hit first and that forwards by the partition mapping (it doesn't handle data itself); 3. Let clients know the partition mapping and connect directly. The hard part is the same in all three—when the partition-to-node mapping changes, who learns of it, and how is the decision-maker told. Many systems rely on a separate coordination service (like ZooKeeper) to hold this cluster metadata and notify subscribers on change (HBase, SolrCloud, older Kafka do this); Cassandra and Riak instead use a gossip protocol among nodes and need no external coordinator; MongoDB uses dedicated config servers plus a mongos routing process.
Partitioning is the foundation of every data system too big for one machine. Setting a shard key in MongoDB, a partition key in Cassandra, partition counts for a Kafka topic, a partition key in DynamoDB—all are choices within this chapter's framework: range or hash, local or global index, will the partition key cause hot spots. The high-frequency interview questions—"how do you choose Kafka partition counts," "what's the difference between Cassandra's partition key and clustering key," "how do you avoid a hot partition / hot key," "what is consistent hashing and how does it differ from modulo," "how do secondary queries work after sharding"—are all answered here. Choosing the wrong partition key is about the most common and least fixable architectural mistake in distributed databases: it dictates how data spreads, where hot spots form, and whether range queries work—and once you're live with data loaded, re-choosing the partition key usually means a massive migration.
(channel_id, bucket) as a compound partition key—bucket being a static time window (about 10 days)—which both groups a channel's messages by time and keeps a busy channel's messages from blowing out a single partition; a textbook case of "compound key + bucketing" against hot spots and unbounded partitions. They later moved storage from Cassandra to ScyllaDB to tame tail latency.Source: Discord Engineering, "How Discord Stores Trillions of Messages" (2023)hash mod N for partitioning: it looks simple but moves nearly all data on scale-out. Use rebalancing-friendly schemes like fixed-count or dynamic partitioning.① Partitioning = split a dataset too big for one machine into non-overlapping chunks, one node each, to scale out storage and throughput; orthogonal to replication, usually stacked.
② Two splits: by key range (sorted, great for range scans, but rising keys cause write hot spots) and by hash (even spread, but loses range queries); Cassandra's compound key takes half of each.
③ Hashing spreads different keys; it can't stop a single hot key—celebrities / viral items need application-level salting, at the cost of merging on read plus bookkeeping.
④ Secondary indexes, two roads: local (write-cheap, read needs scatter/gather across all partitions) and global (read-efficient, write spans partitions and is often async)—essentially, put the cost on reads or on writes.
⑤ Rebalancing: hash mod N is off-limits (moves everything); use fixed partition count / dynamic partitioning / proportional to nodes, moving only whole partitions.
⑥ Fully automatic rebalancing plus automatic failure detection invites cascading failures; production usually keeps a human confirmation gate.
⑦ Request routing must answer "which node holds the key": via a coordination service like ZooKeeper or via gossip between nodes; the mapping changes with rebalancing.
⑧ The partition key is the hardest decision to change: it locks in hot spots, range ability, and scalability—getting it wrong usually means a massive migration, so stress-test before going live.