Why architecture matters here
Utilization skew is not cosmetic; it converts into three concrete production pathologies. First, write failures ahead of capacity: HDFS chooses write targets among nodes with free space, so when the fullest third of the fleet crosses ~95%, the effective write bandwidth of the cluster shrinks toward the remaining nodes long before aggregate capacity runs out — a cluster that is 78% full on average can start throwing could only be replicated to 0 nodes because the placement policy keeps landing on full disks. Second, locality loss: YARN schedules tasks where the data is; when recent data concentrates on a few ingest nodes, those nodes oversubscribe on CPU while the rest of the cluster reads over the network, and shuffle-heavy jobs slow down measurably.
Third, hotspot amplification on failure: when an over-full node dies, re-replication of its outsized block population hammers the cluster harder and longer than the loss of a balanced node would. Skew also stretches decommission times — draining a 90%-full node moves twice the bytes of a 45% one — and makes disk-failure recovery lopsided.
The balancer’s architecture matters because the naive alternative — copy files around with distcp and fix up replication — is both slower and wrong: it churns namespace metadata, breaks in-flight readers, and cannot preserve placement invariants atomically. The balancer instead moves replicas underneath unchanged files: readers never see a missing block, the NameNode’s namespace is untouched, and every individual move either completes and retires the old replica or fails and leaves the original in place. Rebalancing is a data-plane operation, and the design keeps it there.
The architecture: every piece explained
Top row: the brain. The Balancer is not a NameNode service; it is a client process (typically run from an edge node) that speaks two protocols — to the NameNode for the live DataNode report (capacity, used, per-node utilization) and getBlocks() to fetch batches of block metadata from over-utilized nodes; and to DataNodes to command actual moves. The threshold policy classifies every node against the cluster mean: over-utilized (above mean + threshold), above-average, below-average, and under-utilized (below mean − threshold). The default threshold of 10 percentage points defines ‘balanced’; tightening it to 5 doubles the work and the runtime.
Middle row: the move primitive. For each scheduled move, the balancer picks a source (over-utilized node whose utilization it wants to shed), a target (under-utilized node with room), and a block whose relocation does not violate placement: the target must not already hold a replica, and rack policy — replicas spread across at least two racks — must survive the move. The transfer itself uses a proxy: any DataNode already holding a replica (ideally rack-local to the target) streams the block to the target via replaceBlock; on success the NameNode is told to delete the surplus replica from the source. The source node thus loses bytes without ever sending them.
Bottom row: the governors. Every DataNode caps balancer traffic at dfs.datanode.balance.bandwidthPerSec (live-adjustable via dfsadmin -setBalancerBandwidth), and concurrent-mover limits (dfs.datanode.balance.max.concurrent.moves) bound per-node parallelism. The Disk Balancer is a separate intra-node tool: hdfs diskbalancer -plan node computes a JSON plan of volume-to-volume block moves inside one DataNode, -execute runs it locally — the answer to one new 16 TB drive sitting empty next to nine full ones, which the cluster-level balancer cannot see because it reasons per-node, not per-volume.
Two design details reward attention. First, heterogeneity handling: utilization is computed as a percentage of each node’s capacity, so a 40 TB node and a 100 TB node are both ‘balanced’ at 70% — which equalizes fullness, not block count, and deliberately gives bigger nodes proportionally more data (and more read traffic; plan NIC capacity accordingly). Second, the balancer is storage-type aware: with tiered storage (SSD/DISK/ARCHIVE policies), moves are computed per storage type, and its sibling tool, the Mover, handles the different job of migrating blocks whose storage policy changed — balancer evens out utilization within a tier; Mover enforces policy across tiers. Confusing the two leads to running the wrong tool and wondering why nothing converges.
End-to-end flow
An operator adds six empty DataNodes to a forty-node cluster averaging 72% and runs hdfs balancer -threshold 10. Iteration one begins: the balancer fetches the node report, computes the new mean (~62%), and classifies — the six new nodes are under-utilized, a dozen ingest-heavy veterans are over-utilized. It pairs movers preferring rack-local transfers, requests block lists from sources via getBlocks() (2 GB of metadata at a time), and filters each candidate block against the constraints: target lacks a replica, rack count preserved, block not currently under-replicated or being written.
Scheduling proceeds up to the per-iteration byte budget — each source-target pair moves at most 10 GB per iteration (dfs.balancer.max-size-to-move) — and dispatches moves in parallel across pairs. Each move: proxy DataNode streams the block to the target, target commits it and reports the new replica to the NameNode, NameNode marks the source replica excess and schedules its deletion via the next heartbeat. Failures (target went busy, block deleted mid-flight) are logged and skipped — the original replica is still there, so nothing is at risk. At iteration’s end the balancer prints bytes moved, bytes remaining, and re-fetches the node report; utilization numbers shift a fraction of a percent per pass.
The loop repeats — replan, reclassify, move — for hours or days, exiting when every node lands within the threshold band, when an iteration finds no legal moves, or when the operator stops it (it is safely resumable; state is recomputed from the live cluster each iteration). On this cluster, pulling six nodes from 0% to ~55% means moving roughly 40 TB × 6 of replica data through the proxy path; at the default 100 MB/s per-node cap the rebalance is a multi-day background hum, which is precisely the intent — correctness and invisibility over speed.
Two timing subtleties shape real runs. The NameNode enforces a cool-down between getBlocks calls and caps balancer RPC load so planning traffic cannot crowd out client RPCs on a busy namespace — on RPC-saturated clusters, the balancer’s planning phase itself slows down, which is by design. And deletions lag moves: the source replica is only reclaimed after the NameNode’s invalidation reaches the source DataNode on a heartbeat, so mid-rebalance the cluster transiently holds extra copies of in-flight blocks; on very full clusters this matters — a cluster at 92% average has little room for transient duplication, which is one more reason to balance before the fleet gets critically full, not after.