Hotshard
Distributed data movement

Rebalancing & Data Movement

You added a node. How does the data move to it without taking the system down?

A cluster is never the size you first planned. Traffic grows and you add machines; a machine dies and you're down one; a node runs hot and you want to spread its load. Every one of these changes the map of which node owns which data β€” and a map that changed means data has to physically move to match it. Rebalancing is that movement: shifting data between nodes so the load stays even, while the system keeps serving live reads and writes the whole time. The trap is doing too much of it. Move more than you have to and you flood the network with copying that starves the real traffic. This page is about moving the least data possible, and moving it without downtime. It builds on two things you should know first: consistent hashing (/chash), which decides how little has to move, and sharding (/sharding), which decides how a key maps to a node. Here we take those as given and focus on the mechanics neither one covers β€” how the move actually happens.

~14 min read

Start here: a changed map means data must move#

TL;DRthe 30-second version
  • When the set of nodes changes, the map of who-owns-what changes with it, so data must move to match. Rebalancing is that movement, done while the cluster keeps serving traffic.
  • The naive move is to re-hash every key with the new node count (hash(key) mod N). That remaps almost everything and triggers a copy storm that saturates the network and degrades live requests β€” the same disaster consistent hashing exists to avoid (/chash).
  • Two placement choices keep the move small: a fixed set of many partitions that you shuffle whole between nodes, or virtual nodes so each machine owns many small ring slices (/sharding covers the strategies).
  • The actual migration is a state machine: assign the data to its new owner, stream a copy while dual-writing new changes, verify, cut over reads, then drop the old copy β€” all without a moment of downtime.
  • Throttle the copy so it never starves live traffic, and be careful with automatic rebalancing: a cascading auto-rebalance under load can turn a small problem into an outage.

You're running a 4-node key-value store and it's getting full, so you add a fifth node. The moment it joins, the ownership map has to change β€” the new node has to be responsible for some keys, or it's just an idle machine. But the keys it should now own are currently sitting on the other four. Nothing routes to the new node until those keys are actually copied onto it. So a membership change isn't done when the node boots; it's done when the data finishes moving.

The obvious way to decide what moves is to re-run the placement formula with the new node count. If you were routing with server = hash(key) mod 4, you now route with hash(key) mod 5. Simple β€” except that changing the modulus changes the answer for almost every key. A key that hashed to node 1 under mod 4 almost certainly lands on a different node under mod 5. Roughly 4 out of 5 keys are suddenly in the wrong place and have to be copied.

Why moving too much is its own outageSay each node holds 500 GB. Re-hashing everything means moving on the order of 1.5–2 TB across the cluster at once. That copy traffic shares the same network cards and disks as your live reads and writes, so while the migration runs, real requests slow to a crawl and latencies spike. You didn't just add a node β€” you took the cluster down to move data it mostly didn't need to move. The goal of rebalancing is to move the small slice that genuinely has to move, and nothing else.

You've seen the fix for what to move: consistent hashing makes a membership change touch only about 1/N of the keys instead of nearly all of them, and the sharding page compares the strategies that get you there. The simulators make it concrete β€” the ring's minimal remap on join and leave (/chash/sim), and the resharding move-count counterfactual that shows plain hash-mod moving almost everything versus a stable scheme moving a sliver (/sharding/sim). This page assumes that battle is won. The question left over is the hard one in practice: once you know which slice must move, how do you move it while the system is live?

First, make the move small: partitions and vnodes#

Before the mechanics of moving, one placement decision decides how painful the whole thing will be. The instinct is to map keys straight onto nodes, so the unit you move is 'this node's keys.' That's the trap: it ties the amount of data that moves to the raw arithmetic of node counts, and it makes a node join or leave a big, lumpy event. The better designs put a layer of fixed-size units in between, so a rebalance is just handing a few of those units to a different owner.

The first design is a fixed number of partitions. You pick a partition count once β€” larger than you'll ever have nodes, say 256 partitions for a cluster you expect to grow to a few dozen machines β€” and split the whole key space into that many buckets up front. Each node then owns a handful of whole partitions. When a new node joins, it doesn't get a re-hash; it just claims a few existing partitions from the nodes that have the most, and only those partitions' data moves. The partition count never changes β€” only which node holds each partition. This is the model Riak and Elasticsearch use.

The second design is virtual nodes: instead of one position on the hash ring, each physical machine is given many ring positions (Cassandra historically used 256 per node). Each position owns a small slice of the ring, so a machine owns many scattered slices that add up to its fair share. This is a consistent-hashing detail covered in depth on the ring page (/chash) β€” what matters for rebalancing is the consequence. When a node joins, it takes over one slice from each of many existing nodes, so the incoming data is streamed from many sources in parallel rather than dragged off one overloaded neighbor. When a node dies, its slices are absorbed by many successors, so no single node inherits a crushing load. Both designs share the same idea: many small units, so a membership change reshuffles a fair, small handful of them.

16 fixed partitions (P0–P15) across a cluster, before and after adding node D

Before (3 nodes)AP0–P4BP5–P9CP10–P15
After (add D)AP0–P3BP5–P8CP10–P13DP4,P9,P14,P15
Fixed partitions: a join claims whole partitions, nothing is re-hashed
The partition is the unit of movementIn both designs the thing that moves is a whole partition (or ring slice), never an individual key re-computed from scratch. That's what makes a rebalance cheap to reason about: you move N partitions from over-loaded nodes to the newcomer, and you can count, throttle, and track the move partition by partition. A partition is small and self-contained, so streaming one is a bounded, resumable job.

The live migration: assign, stream, dual-write, cut over#

Now the core of the page. You've decided partition P moves from node A to node D. You cannot just flip the map and point reads at D β€” D doesn't have the data yet. And you cannot stop writes to P while you copy it, because that's downtime. The move has to happen underneath live traffic, so it runs as a careful sequence where A keeps serving P the entire time until D is provably ready.

  1. Assign: the map is updated to say D is the new owner of P, but marked as still catching up. A stays the active owner and keeps serving all reads and writes for P.
  2. Stream: D pulls a bulk copy of P's current data from A in the background, throttled so it doesn't saturate the link. This is the big, slow part β€” it can take minutes to hours for a large partition.
  3. Dual-write: from the moment streaming starts, every new write to P is applied on both A and D. Without this, changes that land during the long copy would be lost β€” D's bulk snapshot would already be stale by the time it finished.
  4. Verify: D confirms it has the full partition β€” the streamed snapshot plus the dual-written changes. A checksum or a Merkle-tree comparison (/merkle) proves the two copies match before anything switches.
  5. Cut over: the map flips P's active owner to D. New reads and writes now route to D. This is the only instant that matters to a client, and it's atomic β€” a pointer change, not a data move.
  6. Clean up: A drops its now-orphaned copy of P and stops dual-writing. The space is reclaimed. The migration of P is done.
Acurrent ownerDnew owner
assign β€” D marked owner-in-waiting, A still serves P
stream bulk copy of P (throttled)
every live write to P is applied on BOTH nodes
D: snapshot + changes received
verify β€” checksum / Merkle compare, copies match
cut over: map flips, reads/writes β†’ D
A drops its copy, stops dual-writing
Moving partition P from A to D, with no downtime

The shape to remember: copy in the background, write to both during the copy, switch atomically at the end, then delete the old copy. The old owner is the safety net the whole time β€” if D fails mid-stream, you abandon the move and A never stopped serving. Nothing is committed until the verify step proves D is complete, and the cut-over is a cheap pointer flip rather than a data operation. Every node also has to learn about the new map; that's a job for the gossip layer (/gossip/sim), which spreads the updated ownership around the cluster so every client routes to D after the switch.

Go deeperUnder the hood: catching up the writes that land mid-copy

Dual-writing is the clean way to catch changes during the copy, but there's a subtlety: the very first writes can arrive before D has even created the partition locally. Systems handle the overlap in one of two ways. Some do a strict handoff β€” D buffers or logs incoming writes for P and applies them after the bulk snapshot lands, so order is preserved. Others lean on the replication layer's own repair: stream a point-in-time snapshot, then run anti-entropy (/read-repair-ae) to reconcile anything that changed after the snapshot, using Merkle trees to find just the differing ranges rather than re-copying the partition.

Related but different is hinted handoff, which handles a node being temporarily down rather than a planned move. If D is briefly unreachable during normal operation, the coordinator stores a 'hint' β€” the write it couldn't deliver β€” and replays it when D returns. It keeps a short outage from causing lost writes, but it is not a substitute for a full migration: hints are bounded and expire, so a node that's down long enough still needs a real rebuild via streaming.

The other advanced case is a partition that's too hot or too big β€” one partition taking a disproportionate share of traffic. Dynamic-partitioning systems (HBase, DynamoDB) split that partition into two at a chosen key boundary and hand one half to another node, and merge two cold adjacent partitions back together when they shrink. Split and merge are the same migration mechanics applied to a partition boundary instead of a whole partition: copy, verify, cut over.

The numbers: how much moves, and how long it takes#

Two quantities decide how a rebalance feels: how much data has to move, and how fast you're willing to move it. The first is set by the placement scheme; the second is a throttle you choose.

  • How much moves: with a stable scheme (consistent hashing or fixed partitions), adding the (N+1)-th node moves about 1/(N+1) of the data β€” the newcomer's fair share β€” and nothing else churns. Adding a 5th node to a 4-node cluster moves ~1/5 of the data. Under plain hash-mod it would be ~4/5. That gap is the whole reason the placement layer exists.
  • How fast: the copy is deliberately rate-limited. Cassandra's stream throughput defaults to 200 megabits per second per node (about 25 MB/s), tunable live with nodetool setstreamthroughput. The cap exists so streaming never eats the bandwidth live traffic needs.
  • How long: move time is just data Γ· rate. Streaming a 100 GB partition at 25 MB/s takes about an hour. Raising the throttle finishes sooner but steals more bandwidth from live requests; lowering it protects latency but drags the migration out. That dial is the core operational trade-off.
PredictYou add one node to a 9-node cluster where each node holds 400 GB, using consistent hashing. Roughly how much data moves, and where does it come from?

Hint: The newcomer's fair share is 1/(N+1) of the total β€” but with vnodes, which nodes does it pull from?

The new node should end up owning about 1/10 of the data. The total is 10 nodes Γ— 400 GB = 4 TB, so its fair share is ~400 GB β€” that's roughly how much moves onto it, and almost nothing else moves. Where it comes from depends on the scheme: with a single ring position the newcomer would drag its whole ~400 GB off one unlucky neighbor, hammering that one node; with virtual nodes it takes a small slice from each of many existing nodes, so it streams ~400 GB in parallel from all of them and no single source is overloaded. Same total moved, very different blast radius β€” which is exactly why production clusters use vnodes or many fixed partitions instead of one position per node.

The move is resumable, not atomicA whole rebalance isn't one transaction β€” it's many independent partition migrations, each its own copy-verify-cutover. That's a feature: if a node crashes halfway through moving 40 partitions, the 25 already cut over stay put and only the in-flight ones restart. You never have to redo the whole thing, and you can pause a rebalance under load and resume it off-peak.
Automatic vs manual, and how hard to push it

Two knobs turn a correct rebalance into a safe one: whether it triggers itself, and how aggressively it copies. Both are trades, and both have caused real outages when set wrong.

  • Automatic rebalancing is convenient: the cluster notices an imbalance or a failed node and starts moving data on its own, no operator awake at 3 a.m. The danger is a cascade. If a node fails under heavy load and an auto-rebalance immediately starts streaming its data onto already-busy neighbors, the extra copy load can push those neighbors over the edge, which triggers more rebalancing β€” an outage feeding itself. Elasticsearch's default is to wait a grace period after a node leaves before rebalancing, precisely so a quick restart or a brief network blip doesn't kick off a pointless, dangerous data shuffle.
  • Manual (operator-triggered) rebalancing trades convenience for control: the cluster reports that it's unbalanced but waits for a human to run the move, at a time and pace they pick. It's the safer default for large or latency-critical clusters, where you want the copy to happen off-peak and under supervision.
  • The throttle is the other dial. A high stream rate finishes the migration fast but competes hard with live traffic; a low rate protects request latency but leaves the cluster unbalanced (and a new node under-utilized) for longer. There's no universal right answer β€” you tune it to how much headroom the live workload has.
The one-line tradeRebalancing is always latency-now versus balance-soon. Every byte you copy to fix the balance is bandwidth you took from live requests. The art is moving the minimum data, at a rate the live workload can spare, ideally when you choose rather than when the cluster panics.
The three placement models, side by side

Rebalancing behavior is decided by how the system assigns data to nodes. Three models dominate, and they move data differently on a membership change.

Fixed partitionsConsistent hashing + vnodesDynamic partitioning
Partition countFixed at cluster creationFixed per node (vnode count)Grows/shrinks by split & merge
What moves on a joinWhole partitions reassignedRing slices from many nodesSplit halves handed off
Handles data skewPoorly β€” partitions are equal-sized by countPoorly β€” balances key count, not popularityWell β€” hot partition splits
Operational simplicitySimple β€” count never changesSimple, but vnode tuning mattersComplex β€” dynamic split/merge logic
Used byRiak, Elasticsearch, CouchbaseCassandra, DynamoDB (ring)HBase, DynamoDB (adaptive), Bigtable

The rule of thumb: fixed partitions are the simplest to operate when your data is fairly uniform and you know your rough growth ceiling; consistent hashing with vnodes spreads a join or leave across many nodes for gentle failover; dynamic partitioning is what you reach for when a few partitions run hot and you need the system to split them without a full reshard. Many production systems blend them β€” DynamoDB uses a consistent-hash ring and also adaptively splits hot partitions.

How real systems rebalance
  • Riak fixes the number of partitions at cluster creation (default 64, always a power of two) and never changes it. Adding a node hands it a share of existing partitions off the ring; only those partitions' data streams over. The partition count is a permanent capacity-planning decision made up front.
  • Elasticsearch fixes an index's shard count at creation and can't change it in place β€” to add shards you use the Split API (or reindex), which builds a new index. Within a fixed shard count it moves whole shards between nodes to balance, and delays rebalancing after a node leaves so a brief outage doesn't trigger a needless shuffle.
  • Apache Cassandra uses virtual nodes (num_tokens, historically 256, now commonly 16 with a smarter allocation algorithm) so a join streams data in parallel from many nodes. After a node joins and takes over ranges, the old owners keep their now-orphaned copies until you run nodetool cleanup to reclaim the space β€” a manual step people forget. Stream rate is capped by stream_throughput_outbound_megabits_per_sec (default 200).
  • HBase splits a region (its partition) automatically when it grows past a size threshold, and the balancer moves regions between region servers to even out load β€” dynamic partitioning in production.
  • Amazon DynamoDB descends from the Dynamo ring design and adaptively splits partitions that get hot or large, moving data behind the scenes so a single hot key's partition doesn't bottleneck β€” rebalancing the customer never sees.
Pitfalls & gotchas
Why not just re-hash everything when the cluster changes? It's simpler.

Because plain hash(key) mod N remaps almost every key when N changes, so 'simple' means copying nearly the entire dataset at once β€” a network-saturating storm that degrades live traffic and can take the cluster down. That's the exact failure consistent hashing (/chash) and fixed partitions exist to avoid: they move only the ~1/N slice that genuinely has to move.

How is there no downtime if the data is mid-copy?

The old owner keeps serving the partition the entire time. The new owner streams a copy in the background while both nodes apply new writes (dual-write), and only after a verify step confirms the copies match does the map flip β€” atomically β€” to the new owner. The client's request always has a live owner to hit; the only instant of change is a pointer flip, not a data move.

What actually goes wrong with automatic rebalancing?

Cascades under load. A node fails while the cluster is busy, auto-rebalance immediately streams its data onto neighbors that are already near capacity, the copy load tips those neighbors over, and their failure triggers still more rebalancing. A small problem becomes a spreading outage. The guards are a grace period before rebalancing (so a quick restart doesn't trigger it), a throttle, and often requiring an operator to start the move.

The new node joined but disk usage on the old nodes didn't drop. Why?

Many systems don't automatically delete the data a node no longer owns after a range moves β€” the old copy lingers until you explicitly reclaim it (in Cassandra, nodetool cleanup). Until then the moved data exists in two places. Forgetting the cleanup step is a classic way to run nodes out of disk right after a 'successful' rebalance.

QuizA node in a busy 6-node Cassandra cluster crashes at peak. The cluster is configured to auto-rebalance immediately with the stream throttle set very high to 'finish fast.' Moments later two more nodes fall over. What most likely happened?

  1. The hash function was weak and reshuffled every key.
  2. The immediate high-rate rebalance dumped the dead node's data onto already-busy neighbors, and the extra copy load tipped them over β€” a cascade.
  3. Consistent hashing doesn't work with 6 nodes.
  4. The nodes ran nodetool cleanup and deleted live data.
Show answer

The immediate high-rate rebalance dumped the dead node's data onto already-busy neighbors, and the extra copy load tipped them over β€” a cascade. β€” This is the auto-rebalance cascade. At peak load the neighbors were already near capacity; an immediate rebalance at a high stream rate piled a large copy workload on top of the live traffic they were already struggling with, pushing them past their limit. Their failure then triggered more rebalancing β€” the outage feeding itself. The fixes are exactly the guards real systems ship: a grace period before rebalancing kicks in, a throttle low enough that copying never starves live requests, and often a human deciding when to move data. A weak hash or consistent hashing itself isn't the problem here; the trigger policy and the rate are.

In an interview

Rebalancing comes up the moment you say 'and then we add a shard' in a design. The interviewer wants to hear that you know adding a node isn't free β€” data has to move β€” and that you know how to move it without an outage. Lead with the two-part answer: keep the move small, and make it live.

  • Start with why naive re-hashing is wrong: hash mod N remaps almost everything on a resize, a copy storm that degrades live traffic. Name the fix in one clause β€” consistent hashing or fixed partitions move only ~1/N of the data β€” and don't re-derive the ring; reference it and move on.
  • State the placement choice that keeps failover gentle: virtual nodes (or many fixed partitions) so a join pulls from many sources and a failure spreads across many successors, instead of one neighbor inheriting everything.
  • Walk the live migration as a sequence: assign the new owner, stream a copy in the background, dual-write during the copy so nothing is lost, verify the copies match, cut over atomically, then drop the old copy. The old owner serving throughout is why there's no downtime.
  • Name the throttle and the trade: copy rate is latency-now vs balance-soon; you move the minimum data at a rate the live workload can spare. Cite a real number β€” Cassandra's 200 Mbit/s default stream cap.
  • Flag the auto-rebalance cascade as the operational trap, and the forgotten-cleanup gotcha (moved data lingering on the old node until reclaimed). Naming those two signals you've actually operated one of these, not just read about it.
PredictThe interviewer says: 'Your design shards user data by consistent hashing. A node dies during Black Friday traffic. Walk me through what happens and what you'd worry about.'

Hint: Separate two things: serving the dead node's keys right now (replicas) versus rebuilding its data (a throttled migration you don't have to do at peak).

First, correctness: the dead node's ring slices are already replicated elsewhere (you'd have replication factor β‰₯ 3), so reads and writes for its keys immediately fail over to the surviving replicas β€” no data lost, brief blip at most. Then the worry: rebuilding the lost replica means streaming a full copy of that node's data onto a replacement or onto the successors, and doing that at peak traffic is dangerous. If it's an automatic, unthrottled rebalance you risk a cascade β€” the copy load on already-saturated neighbors tips them over too. So the strong answer is: let replication cover the reads immediately, but hold or heavily throttle the rebuild until traffic subsides, or cap the stream rate so the copy can't starve live requests. Mention that vnodes help here because the dead node's load and its rebuild spread across many nodes rather than crushing one. That framing β€” failover is instant via replicas, but the data-movement rebuild is the thing you schedule carefully β€” is what separates a strong answer from 'consistent hashing handles it.'

References & further reading
References

Feedback on this topic β†’