Partitioning
Key-range vs hash partitioning, hot spots, local vs global secondary indexes, rebalancing strategies and request routing
Replication copies the whole dataset to every node, which stops working when the dataset or write throughput exceeds one machine. Partitioning (sharding) breaks the data into pieces so that each record belongs to exactly one partition, and partitions live on different nodes of a shared-nothing cluster. A query touching one partition is served by one node; a query touching many runs in parallel. Partitioning is how a database scales horizontally.
The vocabulary is fragmented: a partition is a shard in MongoDB, Elasticsearch and SolrCloud, a region in HBase, a tablet in Bigtable, a vnode in Cassandra and Riak, a vBucket in Couchbase. The questions are the same everywhere: how to decide which partition a record belongs to, how to index data spread across partitions, how to move partitions when nodes join or leave, and how a request finds the right node.
In one minute
- Partitioning and replication are combined: each partition has replicas on several nodes, and in a leader-based system a node leads some partitions and follows others.
- The goal is to spread data and load evenly. A skewed partition with disproportionate load is a hot spot, and it makes a cluster no faster than one node.
- Key-range partitioning keeps keys sorted so range scans are cheap but makes sequential keys hot. Hash partitioning spreads keys uniformly but destroys range scans. Cassandra's compound keys hash the first column and sort by the rest.
- A single extremely hot key cannot be fixed by the database; the application splits it with a random suffix and pays on reads.
- Secondary indexes are document-partitioned (local: cheap writes, scatter/gather reads) or term-partitioned (global: targeted reads, slow asynchronous writes).
- Never assign by
hash mod N. Rebalance with a fixed number of partitions (Riak, Elasticsearch), dynamic splitting (HBase, MongoDB) or partitions proportional to nodes (Cassandra). - Fully automatic rebalancing plus automatic failure detection is a recipe for cascading failure; many systems propose a plan and let an operator approve it.
- Request routing is service discovery, solved with a coordination service (ZooKeeper, etcd), a gossip protocol (Cassandra, Riak) or partition-aware clients.
Partitioning and replication
Partitioning and replication are orthogonal and almost always combined. Each partition is stored on several nodes for fault tolerance, and each node holds several partitions. With leader-based replication every partition has its own leader, so a node leads some partitions and follows others; the schemes of Chapter 5 apply per partition and the rest of this chapter ignores them. Note that "partition" here is unrelated to the network partition (netsplit) of Chapter 8.
Partitioning of key-value data
The target is to spread data and query load evenly: ten nodes should handle ten times the data and throughput of one. Unfair distribution is skew; a partition with disproportionately high load is a hot spot, and in the worst case nine of ten nodes idle. Random placement would spread data perfectly but make reads untargetable, so both schemes below locate a record by its key.
Partitioning by key range
Assign each partition a contiguous range of keys, like the volumes of a paper encyclopedia. Knowing the boundaries tells you which partition holds a key. Boundaries need not be evenly spaced: the data is not evenly distributed, so a volume for "A–B" and one for "T–Z" may hold similar amounts. Boundaries are chosen by an administrator or by the database.
Within a partition keys stay sorted (Chapter 3's SSTables), so range scans are efficient and the key acts as a concatenated index for fetching related records together. Bigtable, HBase, RethinkDB and MongoDB before 2.4 use this. The downside is hot spots from access patterns: sensor data keyed by timestamp sends every write of the day to one partition. The fix is to prefix the key with something that spreads writes, such as the sensor name, so each sensor's time range stays contiguous; a query across all sensors then becomes one range query per sensor.
Partitioning by hash of key
To remove skew from the key distribution, partition by a hash of the key, which turns skewed input into uniform output. Cryptographic strength is unnecessary: Cassandra and MongoDB use MD5, Voldemort uses Fowler–Noll–Vo. Avoid language built-ins such as Java's Object.hashCode() or Ruby's Object#hash, which can return different values for the same key in different processes. Each partition owns a range of hash values; boundaries may be evenly spaced or pseudorandom.
A note on consistent hashing: Karger et al.'s 1997 technique distributes load across an internet-wide cache (a CDN) with randomly chosen boundaries and no central control. It rarely fits databases in that form, so the book prefers the plainer term hash partitioning; in practice "consistent hashing" is used loosely for any hash scheme where adding a node moves few keys.
The cost of hashing is losing efficient range queries: adjacent keys are scattered everywhere. MongoDB with hashed sharding sends range queries to all partitions; Riak, Couchbase and Voldemort do not support range queries on the primary key. Cassandra's compound primary key is the compromise: only the first column is hashed to pick the partition, the remaining columns sort rows within it. With key (user_id, update_timestamp), all of one user's updates sit in one partition in time order, so "this user's last 24 hours" is one scan. This is the standard model for one-to-many relationships.
| Property | Key-range partitioning | Hash partitioning |
|---|---|---|
| Range queries | Efficient | Scatter to all partitions or unsupported |
| Sequential keys | Hot spots | Spread uniformly |
| Boundaries | Adapted to data distribution | Hash ranges, even or pseudorandom |
| Key locality | Related keys together | Related keys scattered |
| Examples | Bigtable, HBase, RethinkDB | Cassandra, Riak, Voldemort, MongoDB hashed |
Skewed workloads and relieving hot spots
Hashing cannot help when one key is hot: a celebrity's post attracting millions of writes still hashes to one partition. Databases do not compensate for this automatically today, so the application must. For a known hot key, append a random two-digit number to spread it across 100 keys and 100 partitions. Reads then fetch all 100 and combine them, and the application must track which keys are split, so it is done only for the few genuinely hot keys.
Partitioning and secondary indexes
Everything above assumes access by primary key. Secondary indexes (all red cars, all articles mentioning a word) are the bread and butter of relational databases, common in document databases and the whole point of Elasticsearch and Solr. They do not map neatly to partitions, because an index entry does not identify a record uniquely. Two approaches exist.
Partitioning secondary indexes by document
Each partition maintains its own index covering only its documents: a local or document-partitioned index. A write to document 893 updates indexes only in document 893's partition. Reads are the problem: red cars exist in every partition, so a query for color:red must be sent to all partitions and the results merged. This scatter/gather suffers tail latency amplification, being as slow as the slowest partition. It is nonetheless the most common design: MongoDB, Riak, Cassandra, Elasticsearch, SolrCloud and VoltDB all use it, and vendors advise structuring the partitioning so secondary-index queries hit one partition where possible.
Partitioning secondary indexes by term
Alternatively, build a global index over all partitions and partition the index itself, by the indexed value rather than by document: red cars from every data partition appear in the index partition for colours "a" through "r". This is a term-partitioned index (from full-text search, where terms are words). The index can be partitioned by the term itself (enabling range scans on the indexed value) or by its hash (more even load). Reads become efficient, hitting one index partition. Writes become slower: a document update may touch several terms in several index partitions on different nodes, which would require a distributed transaction to keep in sync. In practice global indexes are updated asynchronously: DynamoDB documents that its global secondary indexes usually propagate within a fraction of a second but may lag under infrastructure faults. Riak Search and Oracle's data warehouse (which lets you choose local or global) are other examples.
| Document-partitioned (local) | Term-partitioned (global) | |
|---|---|---|
| Write cost | One partition | Several partitions, typically asynchronous |
| Read cost | Scatter/gather across all partitions | One index partition |
| Index freshness | Can be synchronous | Eventually consistent |
| Examples | MongoDB, Cassandra, Elasticsearch, SolrCloud, VoltDB | DynamoDB GSIs, Riak Search, Oracle optional |
Rebalancing partitions
Throughput grows, datasets grow, machines fail: each change moves data and requests between nodes. This rebalancing should leave load fairly shared, keep accepting reads and writes while in progress, and move no more data than necessary.
How not to do it: hash mod N
Assigning key k to node hash(k) mod N fails because changing N moves most keys: hash 123456 goes to node 6 of 10, node 3 of 11, node 0 of 12. Adding one node reshuffles nearly everything. Any scheme where a key's location depends on the node count has this flaw; the strategies below decouple key-to-partition from partition-to-node.
Fixed number of partitions
Create many more partitions than nodes and give each node several: 1,000 partitions on 10 nodes is about 100 each. A new node steals a few partitions from every existing node; a departing node's partitions are redistributed. Only whole partitions move and the key-to-partition mapping never changes; during a transfer the old assignment keeps serving. Heterogeneous hardware is handled by giving stronger nodes more partitions. Riak, Elasticsearch, Couchbase and Voldemort use this.
The catch is choosing the number up front, since it is rarely changed later. Too many partitions and per-partition overhead (management, replication, recovery) dominates; too few and partitions grow huge, making rebalancing and recovery slow. Hundreds to low thousands is a common range; when the dataset size is highly variable the choice is hard.
Dynamic partitioning
For key-range partitioning fixed boundaries are dangerous: wrong boundaries put all data in one partition. HBase and RethinkDB therefore split partitions dynamically: when one exceeds a configured size (HBase's default region size is 10 GB) it splits in two; when it shrinks below a threshold it merges with a neighbour, like the top level of a B-tree. Each partition lives on one node and a node holds many; after a split one half can move to another node (HBase does this through HDFS). The partition count adapts to data volume: small overhead for small data.
The caveat is the empty database: with no boundary information it starts with a single partition, so all writes hit one node until the first split. HBase and MongoDB support pre-splitting an initial set of partitions, which requires knowing the expected key distribution. Dynamic partitioning works for hash as well as key-range partitioning (MongoDB since 2.4).
Partitioning proportionally to nodes
A third option, used by Cassandra and Ketama, fixes the number of partitions per node (Cassandra's default is 256). Partitions grow with data while the node count is constant and shrink again when nodes are added, keeping partition sizes stable. A joining node picks a fixed number of existing partitions at random and takes half of each. Random splits are unfair individually but even out across 256 partitions per node. Random boundaries require hash partitioning, making this the closest to original consistent hashing. Cassandra 3.0 added an alternative algorithm that avoids unfair splits.
| Strategy | Partition count | Partition size | Partitioning type | Used by |
|---|---|---|---|---|
| Fixed number | Set at creation | Grows with data | Key-range or hash | Riak, Elasticsearch, Couchbase, Voldemort |
| Dynamic | Grows and shrinks with data | Bounded by split threshold | Key-range or hash | HBase, RethinkDB, MongoDB |
| Proportional to nodes | Fixed per node | Stable as nodes are added | Hash | Cassandra, Ketama |
Operations: automatic or manual rebalancing
Fully automatic rebalancing saves operations work but is unpredictable: moving partitions reroutes requests and shifts lots of data, loading the network and nodes while other requests suffer. Combined with automatic failure detection it is a classic cause of cascading failure: a slow node is declared dead, rebalancing away from it slows the others, more nodes are declared dead, and so on. Couchbase, Riak and Voldemort therefore generate a suggested assignment and require an administrator to commit it. Slower, but a human can decide that a traffic spike is the wrong moment to move a terabyte.
Request routing
A client wanting key "foo" needs to know which IP address and port to contact, and partition assignments change. This is an instance of service discovery, which every networked service must solve.
Three places to put routing knowledge
- Any node. Clients contact any node (through a round-robin load balancer); a node that does not own the partition forwards the request and relays the reply.
- A routing tier. A partition-aware load balancer holds no data and only forwards.
- Partition-aware clients. Clients know the assignment and connect directly.
In every case the hard question is how the deciding component learns about changes. Everyone must agree, or requests go to the wrong node. That is a consensus problem (Chapter 9).
Coordination services and gossip
Many systems delegate cluster metadata to a coordination service such as ZooKeeper (or etcd, Consul). Nodes register themselves; ZooKeeper holds the authoritative partition-to-node map; routers or clients subscribe and are notified of every change. HBase, SolrCloud and Kafka (before KRaft) use ZooKeeper this way, LinkedIn's Espresso uses Helix on top of it, and MongoDB uses its own config servers with mongos routing daemons.
Cassandra and Riak instead use a gossip protocol: nodes exchange cluster state peer to peer, any node accepts requests and forwards them (approach 1), and no external coordination service is needed, at the cost of more complexity inside the database. Couchbase does not rebalance automatically, which simplifies things, and uses a routing tier called moxi. Whatever the approach, clients still need the IP of a node or router; since those change rarely, DNS suffices for that first hop.
Parallel query execution
The above concerns single-key reads and writes and scatter/gather over a secondary index. Massively parallel processing (MPP) warehouses (Teradata, Vertica, Greenplum, SQL-on-Hadoop engines) run complex analytical queries with joins, grouping and aggregation by having the optimiser break a query into stages and partitions executed in parallel across nodes. Chapter 10 covers the techniques; partitioning is what makes that parallelism possible.
Design trade-offs
| Decision | Option A | Option B | Choose A when | Choose B when |
|---|---|---|---|---|
| Key assignment | Key-range | Hash | Range scans dominate and keys are not sequential | Uniform load matters; model one-to-many with compound keys |
| Secondary indexes | Local | Global | Writes dominate, queries can be scoped to a partition | Secondary-attribute reads dominate and eventual index consistency is fine |
| Rebalancing | Fixed count | Dynamic splitting | Dataset size is predictable | Size varies by orders of magnitude or key-range is used |
| Trigger | Automatic | Operator-approved | Small cluster, cheap moves, strong monitoring | Large data, cascading-failure risk, mature ops |
| Routing | Coordination service | Gossip | You already run ZooKeeper/etcd and want consistent metadata | You want no external dependency and tolerate eventual metadata |
A sensible interview default: hash-partition on a key that groups the data you query together (user or tenant ID), use many fixed partitions assigned through a coordination service, keep secondary indexes local, then discuss the access patterns that push you off that default.
Interview angle
-
Explain key-range versus hash partitioning and when to use each. Key-range keeps keys sorted per partition, giving cheap range scans and locality, but monotonically increasing keys make one partition hot. Hash partitioning spreads keys uniformly but scatters range queries. Choose key-range when scans dominate and the key is not sequential; choose hash for uniform load, grouping related records with a compound key as Cassandra does.
-
You store time-series sensor data. How do you avoid a hot partition? Keying by timestamp alone concentrates writes on one partition. Lead the key with the sensor ID so writes spread across sensors while each sensor's time range stays contiguous; an all-sensors query becomes one range query per sensor.
-
A celebrity's post gets millions of likes per minute. Does partitioning help? Not by itself: one key hashes to one partition. The application splits the hot key into, say, 100 sub-keys by appending a random number and sums them on read, tracking which keys are split so ordinary keys are not penalised.
-
Compare local and global secondary indexes. A local index lives with its partition's data, so writes touch one partition but queries scatter to all and gather, with tail-latency amplification. A global index is partitioned by indexed value, so queries hit one index partition, but writes may touch several index partitions on other nodes and are applied asynchronously, making the index eventually consistent (as with DynamoDB GSIs).
-
Why is hash mod N bad? The partition depends on the node count, so going from 10 to 11 nodes reassigns almost every key. Good schemes fix the key-to-partition mapping and move whole partitions between nodes.
-
Compare the three rebalancing strategies. Fixed count (Riak, Elasticsearch) is simple but the number must be chosen up front and partitions grow with data. Dynamic splitting (HBase, MongoDB) adapts to data volume but starts with one partition unless pre-split. Proportional to nodes (Cassandra) keeps partition sizes stable as the cluster grows, requires hash partitioning and tolerates slightly unfair random splits.
-
What goes wrong with fully automatic rebalancing? Moving data and rerouting requests loads nodes and network; combined with automatic failure detection, a slow node declared dead triggers rebalancing that slows others until they are declared dead too. Many systems propose a plan and let an operator commit it.
-
How does a client find the node holding a key? Contact any node and let it forward; use a partition-aware routing tier; or make the client partition-aware. The mapping is kept consistent through a coordination service (ZooKeeper for HBase, Kafka, SolrCloud; config servers for MongoDB) or gossip (Cassandra, Riak), and DNS handles the first hop.
Pitfalls and misconceptions
- "Consistent hashing is what databases do." The 1997 ring construction is rarely used as-is; most databases use hash partitioning with explicit ranges and a partition-to-node map.
- "Hashing eliminates hot spots." It spreads distinct keys evenly but does nothing for one hot key, which is an application problem.
- "More partitions are always better." Each has fixed overhead for management, replication and recovery; thousands are fine, millions are not.
- "Secondary indexes behave as on a single node." They become either scatter/gather or an eventually consistent lookup.
- "Range partitioning is bad because of hot spots." Only for monotonically increasing keys; with a good leading key component it offers locality hashing cannot.
- "Partitioning gives linear scaling." Only for single-partition operations; scatter/gather queries can get slower as nodes are added.
- "Rebalancing is instantaneous." It takes as long as copying the data, during which the old assignment keeps serving.
- "A network partition is a data partition." Unrelated concepts sharing a word.
Key terms
| Term | Meaning |
|---|---|
| Partition (shard, region, tablet, vnode) | A subset of the data served as a unit; each record belongs to exactly one |
| Skew | Uneven distribution of data or load across partitions |
| Hot spot | A partition or key with disproportionately high load |
| Key-range partitioning | Assigning contiguous sorted key ranges to partitions |
| Hash partitioning | Assigning ranges of hash values to partitions |
| Consistent hashing | Karger et al.'s hash-ring technique; loosely, any scheme that moves few keys on membership change |
| Compound primary key | The first part selects the partition, the rest sorts within it |
| Document-partitioned (local) index | Each partition indexes only its own records |
| Term-partitioned (global) index | The index is partitioned by indexed value across the cluster |
| Scatter/gather | Querying all partitions and merging results |
| Tail latency amplification | A fan-out request is as slow as its slowest sub-request |
| Rebalancing | Moving partitions between nodes to even out load |
| Fixed number of partitions | Many partitions created up front; nodes gain or lose whole partitions |
| Dynamic partitioning | Partitions split when too large and merge when too small |
| Pre-splitting | Creating an initial set of partitions on an empty database |
| Partitioning proportional to nodes | A fixed number of partitions per node; new nodes split random partitions |
| Cascading failure | Handling one failure overloads other nodes into failing |
| Service discovery | Finding which node currently serves a partition |
| Coordination service | A strongly consistent store (ZooKeeper, etcd) holding cluster metadata |
| Gossip protocol | Peer-to-peer dissemination of cluster state without a coordinator |
Further reading
- DDIA, Chapter 6, "Partitioning". Reference list: https://github.com/ept/ddia-references
- Chang et al., "Bigtable: A Distributed Storage System for Structured Data", OSDI 2006: https://www.usenix.org/legacy/event/osdi06/tech/chang/chang.pdf
- DeCandia et al., "Dynamo: Amazon's Highly Available Key-value Store", SOSP 2007: https://www.allthingsdistributed.com/files/amazon-dynamo-sosp2007.pdf
- Karger et al., "Consistent Hashing and Random Trees: Distributed Caching Protocols for Relieving Hot Spots on the World Wide Web", STOC 1997
- Lamping and Veach, "A Fast, Minimal Memory, Consistent Hash Algorithm": https://arxiv.org/abs/1406.2294
- Hunt et al., "ZooKeeper: Wait-free Coordination for Internet-scale Systems", USENIX ATC 2010: https://www.usenix.org/legacy/event/atc10/tech/full_papers/Hunt.pdf
- Dean and Barroso, "The Tail at Scale", Communications of the ACM, 2013
- Apache HBase Reference Guide, sections on region splitting and pre-splitting