Scaling a Shard Out Without Moving Everything
Adding one node under mod N moves 95% of your data — the classic trap. Runnable demos of mod N vs fixed shards vs consistent hashing, plus the local-vs-global secondary index trade-off: pay on every read, or pay on every write.
You picked a partition key (the sibling post). Now two operational questions decide whether the sharded system is calm or painful: how does it grow, and how do you query by anything other than the key? Both have a right answer and a common trap. Here they are, measured.
Runnable companion:
shard-rebalancing-and-routingon GitHub.make rebalance && make secondaryreproduces every number below (pure Python).
1. Adding a node shouldn't move the whole dataset
Grow the cluster by one node. What fraction of keys change home? The ideal is the new node's fair share (1/new_N) and nothing more:
transition ideal mod N fixed shards consistent
---------------------------------------------------------
4->5 20.0% 79.9% 20.0% 19.0%
10->11 9.1% 90.9% 9.1% 8.8%
20->21 4.8% 95.2% 4.8% 4.3%
hash(key) % N is the trap: because every key's % N changes when N changes, adding one node to a 20-node cluster moves 95% of the data — a full reshuffle for 5% more capacity, i.e. an availability risk dressed up as a routine op. Two schemes avoid it:
- Fixed number of shards — pick a large
Pof logical partitions up front (512 here), spread them over nodes; a new node steals a few whole partitions. Only those keys move. - Consistent hashing — nodes sit on a hash ring with virtual nodes; a joining node adopts only the arcs it now covers.
Both track the ideal line. The rule is blunt: never rebalance with mod N.
2. Routing needs a consistent source of truth
After rebalancing, something must answer "who owns this key now". Put that partition→node map in a consensus store (ZooKeeper/etcd/Raft) so every router agrees and you don't get split-brain (two nodes claiming one key). Gossip-based membership (Dynamo, Cassandra) is the weaker alternative — a natural fit only for leaderless designs that already tolerate eventual consistency. And beware automatic rebalancing married to automatic failure detection: a merely-slow node gets declared dead, its partitions move, the transfer load slows its neighbours, and they're declared dead too — a cascading failure. Rate-limit movement, keep a human in the loop, pre-rebalance before known load spikes.
3. Secondary indexes: pay on read or pay on write
Docs sharded by primary key; to query by another attribute you need a secondary index, sharded one of two ways — with opposite costs (12 shards, 3 indexed fields):
index type read (by attribute) write (one doc)
----------------------------------------------------
local 12 (all shards) 1
global 1 (one shard) 2.8
- Local (document-partitioned): each shard indexes its own docs. Writes cheap (1 shard), always consistent — but an attribute query scatters to all 12 shards and merges. Tail latency = slowest shard, and it doesn't improve with more shards.
- Global (term-partitioned): index sharded by term, so
color=redlives on one shard — reads touch 1. The price is a write fan-out (~2.8 index shards, on other nodes), which forces the update to go asynchronous — and briefly stale:
gap write->read local (sync) global (async)
-----------------------------------------------
0 ticks 0.0% 100.0%
5 ticks 0.0% 28.6%
10 ticks 0.0% 8.2%
Search the tag you just wrote and a global index may not have it yet (DynamoDB's GSIs behave exactly this way); a local index, updated synchronously on the doc's shard, is never stale. Choose by the read/write ratio on that attribute.
What I'd say in an interview
- Never rebalance with
mod N— it moves ~all the data. Fixed logical shards or consistent hashing move only the new node's fair share. - Routing needs a consistent map (ZooKeeper/etcd/Raft) to avoid split-brain; gossip is the leaderless-friendly, weaker option.
- Automatic rebalancing + automatic failure detection can cascade — rate-limit, human-in-the-loop, pre-rebalance.
- Secondary indexes force local vs global — local = cheap writes / scatter reads; global = one-shard reads / write fan-out + async staleness. Pick by read/write ratio.