Partitioning: splitting data across machines
Partitioning by key range
last write: —Bars = share of the last 240 writes. Try: timestamps with key-range partitioning (watch the hot spot walk across partitions as the day goes on), then switch to hash. Then turn on the range scan: key range reads one or two partitions, hash must ask all of them. Finally pick the celebrity: hashing doesn't help a single hot key.
8 hash partitions, one celebrity
Red = writes to the celebrity's key, faint = everyone else. Try: at 40% celebrity traffic one partition takes ~half of all writes. Split the key into 8 or 16 sub-keys (a random two-digit suffix per write) and the load spreads, but now every read of that user's data has to fetch all sub-keys and combine them.
Each partition indexes only its own documents
- color:white → [19, 229]
- make:dacia → [19]
- make:honda → [229]
- color:black → [272]
- color:yellow → [300]
- make:dacia → [300]
- make:ford → [272]
- color:black → [623, 660]
- color:red → [720]
- color:white → [689]
- make:dacia → [689]
- make:ford → [623, 660]
- make:toyota → [720]
- color:blue → [848, 985, 991]
- color:white → [755]
- make:audi → [991]
- make:dacia → [755, 985]
- make:ford → [848]
Try: in local mode, query red cars and see every partition light up (scatter/gather). Switch to global: the query hits one partition, but a write now touches up to three. Add 4 cars and query immediately: the global index is asynchronously updated, so it misses them for a moment.
Rebalancing: how much data moves when you add a node?
rebalancinghash-mod-nfixed-partitionspartitions-proportional-to-nodes600 keys, colored by the node that stores them
Bright, outlined squares = keys that had to move to a different node in the last change. Try: hash mod N, go from 4 to 5 nodes: about 80% of keys move. Switch to fixed partitions and repeat: only whole partitions move, and only the ones the new node takes.
keys per node
Key range partitions that split and merge
0 keysTry: insert random keys from a single partition and watch it split as it fills, the new halves spreading across nodes. Then reset with increasing timestamps: every write lands in the last partition, so one node takes all writes no matter how many splits. Pre-splitting helps random keys at the start, but only if you guessed the key distribution. Delete data to see small neighbours merge.
keys per node
Hash ring: a key belongs to the next token clockwise
Dots = 300 keys at their hash position; outlined dots moved in the last change. Try: with 1 token per node the arcs are wildly uneven, and removing a node dumps its whole arc onto a single neighbour. Go to 16 or 64 tokens per node: shares even out, and a new node takes small slices from everyone.
share of the hash space
Which node do I send this request to?
| partition → | p0 | p1 | p2 | p3 | p4 | p5 |
|---|---|---|---|---|---|---|
| ZooKeeper (authoritative) | A | A | B | B | C | C |
| routing tier's cache | A | A | B | B | C | C |
Green requests went straight to the right node; amber ones hit a stale routing table and needed extra hops. Try: press Play, then Move a partition: until the purple notification arrives, requests for that partition go to the old owner. Set the delay to 0 and the anomaly vanishes. Compare hop counts: a partition-aware client saves the routing tier's two hops.
- takeawayKey-range partitioning keeps keys sorted for cheap scans but risks hot spots. Hash partitioning spreads load and costs you range queries. Neither helps a single hot key, which you have to split in the application.
- takeawayLocal secondary indexes keep writes on one partition and make reads scatter/gather. Global, term-partitioned indexes make reads cheap and writes multi-partition, and are usually updated asynchronously.
- takeawayNever rebalance with hash mod N. Use many fixed partitions, dynamic splitting, or tokens per node, so a new node moves about 1/N of the data, and keep the authoritative assignment in a coordination service so routing can catch up.
Key concepts · 20
book chapter 6- Partition (shard)
A subset of the data such that every record, row or document belongs to exactly one partition.
Partitioning exists for scalability: different partitions live on different nodes, so data volume and query load are spread over many disks and CPUs. It is usually combined with replication, so each partition has copies on several nodes.
MongoDB / Elasticsearch shardHBase regionCassandra vnodeCouchbase vBucket↑ see it in the visualization - Skew
Some partitions get more data or queries than others.
With skew, adding nodes stops helping: with 10 nodes but all load on one, you have the throughput of one node. The goal of partitioning is an even spread of both data and load.
↑ see it in the visualization - Hot spot
A partition with disproportionately high load.
Hot spots come from the access pattern, not only the data size: sequential keys like timestamps send all of today's writes to one partition, and a celebrity's key is hot no matter how you hash it.
↑ see it in the visualization - Partitioning by key range
Each partition owns a contiguous range of keys, like volumes of an encyclopedia; keys are kept sorted within a partition.
Range scans are cheap because neighbouring keys live together. But boundaries must adapt to the data, and access patterns that cluster on adjacent keys (timestamps) create hot spots.
BigtableHBaseRethinkDBMongoDB (before 2.4)↑ see it in the visualization - Partitioning by hash of key
A hash function maps each key to a number; each partition owns a range of hashes.
A good hash spreads even skewed keys uniformly, avoiding hot spots for many patterns. The cost is losing the sort order: a range query has to be sent to all partitions.
Cassandra (Murmur3)MongoDB hashed shard keyVoldemortRiak↑ see it in the visualization - Compound primary key
A key with several columns where only the first is hashed to pick the partition and the rest are kept sorted within it.
A middle ground between hash and range: e.g. (user_id, timestamp) spreads users across partitions but still allows an efficient range scan over one user's updates.
Cassandra partition key + clustering columnsDynamoDB partition + sort key - Splitting a hot key
Append a random number (say two digits) to a very hot key so its writes spread over many sub-keys and partitions.
It relieves the hot partition, but every read must now fetch all sub-keys and combine them, and you need bookkeeping to know which keys are split. Most databases leave this to the application.
↑ see it in the visualization - Document-partitioned (local) index
Each partition maintains a secondary index covering only the documents stored in that partition.
Writes stay on one partition and the index is updated synchronously. But a query by secondary key must be sent to every partition and the results merged.
MongoDBElasticsearchCassandra secondary indexesRiakSolrCloud↑ see it in the visualization - Scatter/gather
Send a query to all partitions in parallel and combine their answers.
The query is only as fast as the slowest partition, so tail-latency amplification makes scatter/gather reads expensive, even when they run in parallel.
↑ see it in the visualization - Term-partitioned (global) index
A secondary index that covers all data but is itself partitioned by the indexed term, separately from the documents.
A read needs only the partition holding that term. A single document write may now touch several partitions, so in practice the index is updated asynchronously and may briefly miss recent writes.
DynamoDB global secondary indexesRiak searchOracle global indexes↑ see it in the visualization - Rebalancing
Moving load from one node to another as data, query rate or the set of machines changes.
Afterwards load should be fair, the database should keep serving reads and writes during the move, and no more data should move than necessary.
↑ see it in the visualization - hash mod N
Pick the node as hash(key) mod number-of-nodes.
Don't: when N changes, most keys map to a different node (going from 4 to 5 nodes moves about 80% of keys), making rebalancing very expensive.
↑ see it in the visualization - Fixed number of partitions
Create many more partitions than nodes up front (e.g. 1,000 for 10 nodes); a new node steals whole partitions from existing ones.
The key-to-partition mapping never changes, only partition-to-node does, so moves are minimal. But the count is fixed at setup: too few and partitions grow huge, too many and overhead adds up.
RiakElasticsearchCouchbaseVoldemort↑ see it in the visualization - Dynamic partitioning
Key-range partitions split in two when they exceed a size threshold and merge with a neighbour when they shrink, like a B-tree's top level.
The number of partitions adapts to the data volume. An empty database starts with one partition, so all writes hit one node until the first split; pre-splitting helps if you know the key distribution.
HBaseRethinkDBMongoDB↑ see it in the visualization - Pre-splitting
Configure an initial set of partition boundaries on an empty database.
It avoids the single-partition bottleneck at the start, but only works if you can predict the key distribution in advance.
HBaseMongoDB↑ see it in the visualization - Partitions proportional to nodes
A fixed number of partitions per node; a joining node splits a random selection of existing partitions and takes half of each.
Partition size stays stable as the cluster grows. Random split points can be unfair, but with enough partitions per node (Cassandra used 256) the load averages out.
Cassandra vnodesKetama↑ see it in the visualization - Consistent hashing
Place randomly chosen tokens on a hash ring; a key belongs to the next token clockwise, so adding or removing a node moves only nearby keys.
It was designed for distributed caches. The book notes the name is confusing and the scheme is rarely used as originally described; in databases, think 'hash partitioning with random boundaries'.
DynamoCassandraAkamai CDN↑ see it in the visualization - Request routing
How a client finds the node that holds the key it wants: ask any node, go through a partition-aware routing tier, or use a partition-aware client.
It is service discovery for partitions. The hard part is everyone agreeing on the assignment while partitions move; a stale view sends requests to the wrong node.
Cassandra/Riak gossipMongoDB mongosRedis Cluster MOVED↑ see it in the visualization - Coordination service (ZooKeeper)
A separate, consistent store of the partition-to-node mapping that nodes register with and routers subscribe to.
It gives one authoritative assignment and notifies subscribers when it changes, so routing can catch up. Many systems rely on it instead of building their own consensus.
ZooKeeper (HBase, SolrCloud, Kafka)Helix (LinkedIn Espresso)MongoDB config servers↑ see it in the visualization - Parallel query execution
Analytic (MPP) databases break complex queries into stages that run on many partitions at once.
Key-value style single-partition reads are the simple case; warehouses push joins, filters and aggregations down to partitions in parallel (the topic returns in chapter 10).
TeradataVerticaRedshift