06

Partitioning: splitting data across machines

Replication keeps copies of the same data on several nodes. Once the data or the write load is too big for one machine, you also need to split it up: each record goes to exactly one partition, and partitions are spread across the cluster. The whole game is spreading load evenly. Pick the wrong key-to-partition mapping and one node does all the work while the others sit idle. The same choice decides whether range queries are cheap, how secondary indexes work, how much data moves when you add a machine, and how a request finds the right node.
The same stream of writes, two ways of assigning keys to partitions. Key ranges keep neighbours together, so scans are cheap, but sequential keys pile up on one partition. Hashing scatters keys evenly and gives up their order.

Partitioning by key range

last write: —
key space (timestamp, sorted)partition boundaries cut the key space directly00:0004:0008:0012:0016:0020:00p00%p10%p20%p30%p40%p50%fair share 1/6

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.

hottest partition
0%
fair share 17%
skew (max ÷ mean)
0.0×
1× = perfectly even
range scan touches
—
turn on a scan
writes simulated
0
A compound primary key gets some of both: hash the first column to pick the partition, keep the rest sorted inside it. With (sensor_id, timestamp), writes spread across sensors, and one sensor's readings over a time range are still a single contiguous scan.

Skew you can't hash away: the celebrity key

hot-spothot-key-splitting
Hashing spreads different keys, but every write to the same key still lands on the same partition. When a celebrity posts and millions react, the usual fix is in the application: split the key by adding a random suffix, then pay for it on reads.
split hot key into

8 hash partitions, one celebrity

p00%p10%p20%p30%p40%p50%p60%p70%
user:celebrity → p2

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.

hottest partition
0%
expected 48% · fair 12.5%
partitions holding the hot key
1 / 8
requests per celebrity read
1
one key, one partition
bookkeeping
none
plain keys
A used-car site, partitioned by listing id, that lets you search by color and make. A secondary index doesn't map neatly onto partitions, so you either index each partition separately or partition the index by term.

Each partition indexes only its own documents

partition 0ids 0–249
● #19 Dacia● #229 Honda
local index
  • color:white → [19, 229]
  • make:dacia → [19]
  • make:honda → [229]
partition 1ids 250–499
● #272 Ford● #300 Dacia
local index
  • color:black → [272]
  • color:yellow → [300]
  • make:dacia → [300]
  • make:ford → [272]
partition 2ids 500–749
● #623 Ford● #660 Ford● #689 Dacia● #720 Toyota
local index
  • color:black → [623, 660]
  • color:red → [720]
  • color:white → [689]
  • make:dacia → [689]
  • make:ford → [623, 660]
  • make:toyota → [720]
partition 3ids 750–999
● #755 Dacia● #848 Ford● #985 Dacia● #991 Audi
local index
  • 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.

partitions per read
4
fan out to all; tail latency adds up
partitions per write
1
doc + local index together
index updates in flight
0
stale query results
0 / 0
reads that missed a recent write

Rebalancing: how much data moves when you add a node?

rebalancinghash-mod-nfixed-partitionspartitions-proportional-to-nodes
Ideally, adding a node moves only the data the new node needs to take on, about 1/N of it. Compare three schemes on the same 600 keys.
4 nodes
last change: 4 → 4 nodes

600 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 moved
0%
0 of 600
minimum needed
0%
no change yet
partitions / node
—
no partitions: key → node
max ÷ mean load
1.24×

keys per node

node A
113
node B
186
node C
136
node D
165
Fully automatic rebalancing is risky. Moving partitions is expensive, and combined with automatic failure detection it can cascade: an overloaded node looks dead, load moves away from it, other nodes get overloaded. Many systems keep a human in the loop to approve the plan.

Dynamic partitioning: split when big, merge when small

dynamic-partitioningpre-splitting
Key-range databases can't fix the boundaries up front without knowing the data. Instead a partition splits at its median key once it grows past a threshold, and one half can move to another node.

Key range partitions that split and merge

0 keys
key space → (rect width = key range owned, fill height = keys stored ÷ split threshold, color = node)0

Try: 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.

partitions
1
threshold 40 keys
splits / merges
0 / 0
keys moved
0
shipped to another node
hottest node, last 25 writes
0%
fair 25%

keys per node

node A
0
1 parts
node B
0
0 parts
node C
0
0 parts
node D
0
0 parts
Each node puts some tokens at random points on the hash ring and owns the arc that ends at each of its tokens. Adding a node takes over a few arcs and leaves the rest of the ring alone. More tokens per node means fairer shares.
remove:
tokens per node

Hash ring: a key belongs to the next token clockwise

DBAChash(key) ∈ [0, 2³²)4 nodes × 1 token0 keys moved in last change

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.

largest share ÷ fair
2.32×
keys moved
0%
0 of 300
nodes giving keys
0
on the last add
nodes taking keys
0
on the last remove

share of the hash space

node A
16.1%
node B
0.3%
node C
58.1%
node D
25.6%

Request routing: who has key k right now?

request-routingcoordination-service
Six partitions on three nodes, and a client sending requests. Routing works fine until a partition moves. Then everything depends on how quickly every router hears about it.

Which node do I send this request to?

node Ap0 p1node Bp2 p3node Cp4 p5ZooKeeperrouting tierclient
partition →p0p1p2p3p4p5
ZooKeeper (authoritative)AABBCC
routing tier's cacheAABBCC

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.

requests served
0
avg network hops
0.00
best case 4
misrouted
0%
0 hit the old owner
stale routing entries
0
all views agree
  • 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
  1. 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
  2. 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
  3. 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
  4. 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
  5. 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
  6. 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
  7. 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
  8. 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
  9. 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
  10. 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
  11. 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
  12. 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
  13. 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
  14. 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
  15. 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
  16. 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
  17. 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
  18. 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
  19. 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
  20. 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