HDFS Decommission Codes and Introduction

Background of HDFS decommission

Currently there are 200+ nodes of 1.22PB HDFS3 and fewer than 100 nodes of 480T HDFS. The decommissioning of nodes is slow, migrating only about 150k blocks per hour on average. A node has about 600k blocks on average, and decommissioning one node takes roughly 4 hours.

Difference between node drop and node decommissioning

Node Decommissioning

First, add the node to the decommissioning list. This tells the NameNode and YARN to stop submitting new tasks and stop writing data to it. Then wait for the blocks on that node to be replicated throughout the cluster. During this time the decommissioning node is preferred as the srcNode (data source), because it has no write requests and its load is low. Other nodes copy blocks from this decommissioning node, so its load will be high during this period. Once all blocks have been replicated, the node state becomes Decommissioned, which can be viewed from the NameNode URL interface.

Note that a decommissioning DataNode also starts replicating 10 minutes and 30 seconds later. It does not start earlier just because decommissioning was actively initiated. The NameNode and DataNode always maintain a simple master/slave relationship: the NameNode never actively initiates any IPC call to the DataNode. Any operation the DataNode must cooperate on is returned via the DatanodeCommand carried on the heartbeat reply.

Node Drop

For example, forcibly stopping a DataNode, the physical machine dying (high load drop, sudden network failure, hardware failure, etc.) are all node drops. By default, 10 minutes 30 seconds later (mainly controlled by two parameters) the NameNode will detect that communication with the node is abnormal and mark it as dropped. The NameNode then looks up all blockids for that node (by IP) and the replica-holding machines, and schedules replication via the heartbeat mechanism. At this point the data source for replication is no longer the dropped node itself, but one of the other nodes holding a replica. Replica placement still follows rack awareness and the replica placement policy.

💡 The difference between a node drop and decommissioning is not only the replication data-source selection; the NameNode's under-replicated-block replication strategy is also different (block replication priority is divided into five levels). As an extreme example, decommissioning a single-replica node will not lose data, whereas a drop of a single-replica node really does lose data.

Network storm caused by a node drop

A node carrying tens of TB, or even hundreds of TB, and millions of blocks going offline will produce a large RPC storm. This is especially challenging for the NameNode in large, high-load clusters: it not only impacts production performance but also creates significant risk, particularly for clusters with bandwidth bottlenecks.

Generally, the NameNode detects a DataNode drop after 103s (heartbeat time) + 25min (NameNode check time, parameter dfs.namenode.heartbeat.recheck-interval) = 10min30s. If within those 10min30s the bandwidth stays saturated, RPC requests are delayed, and DataNode-to-NameNode communication degrades, it is easy for other nodes to keep falling offline, forming a vicious cycle. How should this be avoided?

The NameNode maintains a replication priority queue. For blocks with insufficient replicas, the block with only one replica remaining gets the highest replication priority. So in a cluster with replication factor 2, as soon as one replica goes wrong there is only one replica left, which is the highest priority, and it will be replicated in storm mode. If not controlled well, this can easily affect cluster performance or even bring the cluster down. It is therefore generally not recommended to run a cluster with a replication factor of 2.

💡 L1 (highest): blocks at risk of data loss, e.g. 1) blocks with only one replica (especially 2-replica blocks after one node goes down) or blocks with 0 active replicas; 2) single-replica blocks owned by a node currently being decommissioned.

L2: blocks whose actual replica count is far below the configured value (e.g., 3-replica blocks missing 2), i.e., blocks whose replica count drops below 1/3 of the expected value. These blocks are replicated with the second priority. For example, a 4-replica block missing 3 replicas is replicated before a 4-replica block missing only 2.

L3: under-replicated blocks whose shortage is less severe than L2. Third priority.

L4: blocks that already meet the minimum replica count requirement. Their replication need is lower than L2 and L3.

L5: corrupt blocks that currently have an available non-corrupt replica.

Parameters controlling RPC storms caused by node drops

All three parameters live in hdfs-site.xml. Refer to the Apache Hadoop official website for details. Block replication speed is determined by two aspects: the rate at which the NameNode dispatches tasks, and the rate at which DataNodes replicate between themselves. The former can be thought of as the entrance; the latter as the exit.

These parameters are not available in Hadoop 2.7.3.

Ingress Parameters

Control task dispatch from the NameNode side. Changing this parameter requires a restart of the NameNode, not the DataNode.

dfs.namenode.replication.work.multiplier.per.iteration 
default=2

This parameter determines how many blocks the NameNode tells each DataNode to replicate on each heartbeat (3s) when it sends the task list. For example, in a 500-node cluster with this value set to 10, one heartbeat allows the NameNode to tell DataNodes to replicate 10*500=5000 blocks.

If a node drops/decommissions with 800,000 blocks needing replication, how long does it take the NameNode to finish dispatching all replication tasks to DataNodes?

task dispatch time =
  blocks to replicate / (DataNode count * parameter value) * heartbeat interval

Egress Parameters

In contrast to the above, which controls dispatch from the NameNode, the two parameters below control things at the DataNode level. They also require a restart of the NameNode.

dfs.namenode.replication.max-streams
default=2

This parameter controls the maximum number of replication threads on a DataNode. As mentioned above, block replication priority has five levels. This parameter controls replication of blocks other than the highest priority.

dfs.namenode.replication.max-streams-hard-limit
hadoop default=2, cdh default=40

This parameter controls the number of replication streams for blocks of all priorities on a DataNode, including the highest priority. It is typically used together with the two parameters above.

💡 The former parameter controls how frequently a DataNode accepts tasks, while the latter two further cap the maximum parallel threaded network transfer a DataNode can perform at once. The appropriate values depend on cluster size and configuration and cannot be generalized. It is usually simpler to control things from the ingress side. For example, with a 500-node cluster and `dfs.namenode.replication.work.multiplier.per.iteration=10`, a single heartbeat dispatches 5000 blocks. If cluster files are spread across all 500 nodes and each replicates 10 blocks at once (in practice, rack awareness and replica placement policy mean not all nodes participate), and each block is 128MB, then per-node network load is 128 * 10 / 3 = 546 Mb/s. You then need to check whether that exceeds your actual bandwidth bottleneck and whether such large network IO would affect regular task computation. If so, tune this value down.

How to decommission quickly

The essence of making nodes decommission quickly is to improve replica replication speed. This is mainly controlled by the three parameters above: the first controls NameNode task dispatch, the other two control DataNode replication rate, with the precondition that normal production tasks are not affected. The smaller the cluster, the slower decommissioning, because the overall dispatch volume per heartbeat is lower.

For instance, on a 500-node cluster dfs.namenode.replication.work.multiplier.per.iteration=10; on a 50-node cluster this value should be set to 100 so that NameNode dispatch speed is comparable. Tune to your actual cluster size.

<property>
    <name>dfs.namenode.replication.max-streams</name>
    <value>10</value>
</property>
<property>
    <name>dfs.namenode.replication.max-streams-hard-limit</name>
    <value>20</value>
</property>
<property>
    <name>dfs.namenode.replication.work.multiplier.per.iteration</name>
    <value>5</value>
</property>

In a test cluster of 15 HDFS nodes with ~80k blocks each on average, migration reached 63k blocks per hour, a 5x speedup over the initial 12k.

# cdh configuration
dfs.namenode.replication.work.multiplier.per.iteration=4
dfs.namenode.replication.max-streams=20   
dfs.namenode.replication.max-streams-hard-limit=40

Impact

Significantly increases CPU Load, CPU IO Wait, Disk IO Wait, and other resource utilization.

Source Code Analysis

Once the address in dfs.hosts.exclude is written to the configuration, refreshNodes can read the exclude list and act on it.

/**
 * Rereads conf to get hosts and exclude list file names.
 * Rereads the files to update the hosts and exclude lists.  It
 * checks if any of the hosts have changed states:
 */
public void refreshNodes(final Configuration conf) throws IOException {
  refreshHostsReader(conf);
  namesystem.writeLock();
  try {
    refreshDatanodes();
    countSoftwareVersions();
  } finally {
    namesystem.writeUnlock();
  }
}

Node addresses are read via dfs.exclude, the nodes are marked as decommissioning, and a monitor is used to track them.

public void startDecommission(DatanodeDescriptor node) {
  if (!node.isDecommissionInProgress() && !node.isDecommissioned()) {
    // Update DN stats maintained by HeartbeatManager
    hbManager.startDecommission(node);
    // hbManager.startDecommission will set dead node to decommissioned.
    if (node.isDecommissionInProgress()) {
      for (DatanodeStorageInfo storage : node.getStorageInfos()) {
        LOG.info("Starting decommission of {} {} with {} blocks",
            node, storage, storage.numBlocks());
      }
      node.getLeavingServiceStatus().setStartTime(monotonicNow());
      // Start monitoring this node and add it to pendingNodes
      monitor.startTrackingNode(node);
    }
  } else {
    LOG.trace("startDecommission: Node {} in {}, nothing to do.",
        node, node.getAdminState());
  }
}

Before decommissioning, cluster stats are pre-deducted, for example capacityUsed, capacityTotal, and so on.

The monitor’s run() calls processPendingNodes() to take nodes from pendingNodes and decommission them.

/**
 * Move any pending nodes into outOfServiceNodeBlocks to initiate the
 * decommission or maintenance mode process.
 *
 * This method must be executed under the namenode write lock to prevent
 * the pendingNodes list from being modified externally.
 */
private void processPendingNodes() {
  while (!pendingNodes.isEmpty() &&
      (maxConcurrentTrackedNodes == 0 ||
          outOfServiceNodeBlocks.size() < maxConcurrentTrackedNodes)) {
    outOfServiceNodeBlocks.put(pendingNodes.poll(), null);
  }
}

After obtaining nodes, check() is called, which retrieves block information for these nodes and processes the pending replication blocks.

private void check() {
  final List<DatanodeDescriptor> toRemove = new ArrayList<>();

  if (outOfServiceNodeBlocks.size() == 0) {
    // No nodes currently being tracked so simply return
    return;
  }

  // Check if there are any pending nodes to process, ie those where the
  // storage has not been scanned yet. For all which are pending, scan
  // the storage and load the under-replicated block list into
  // outOfServiceNodeBlocks. As this does not modify any external structures
  // it can be done under the namenode *read* lock, and the lock can be
  // dropped between each storage on each node.
  //
  // TODO - This is an expensive call, depending on how many nodes are
  //        to be processed, but it requires only the read lock and it will
  //        be dropped and re-taken frequently. We may want to throttle this
  //        to process only a few nodes per iteration.
  outOfServiceNodeBlocks.keySet()
      .stream()
      .filter(n -> outOfServiceNodeBlocks.get(n) == null)
      .forEach(n -> scanDatanodeStorage(n, true));

  processMaintenanceNodes();
  // First check the pending replication list and remove any blocks
  // which are now replicated OK. This list is constrained in size so this
  // call should not be overly expensive.
  processPendingReplication();

  // Now move a limited number of blocks to pending
  moveBlocksToPending();

  // Check if any nodes have reached zero blocks and also update the stats
  // exposed via JMX for all nodes still being processed.
  checkForCompletedNodes(toRemove);

  // Finally move the nodes to their final state if they are ready.
  processCompletedNodes(toRemove);
}

1. processPendingReplication() calls isBlockReplicatedOk() (the core), which decides whether a block needs reconstruction and pushes information into the replication queue within blockManager.

blockManager.neededReconstruction.add(block,
    liveReplicas, num.readOnlyReplicas(),
    num.outOfServiceReplicas(),
    blockManager.getExpectedRedundancyNum(block));

2. moveBlocksToPending() checks the queue length and the pending-replication limit, creates a block iterator for each node that needs maintenance or decommissioning, and then for every block keeps calling isBlockReplicatedOk() until the replication limit is reached, at which point it stops and exits the loop.

There is an interesting piece of locking logic inside moveBlocksToPending():

Iterator<DatanodeDescriptor> nodeIter =
    Iterables.cycle(iterators.keySet()).iterator();
while (nodeIter.hasNext()) {
  // Cycle through each node with blocks which still need processed
  DatanodeDescriptor dn = nodeIter.next();
  Iterator<BlockInfo> blockIt = iterators.get(dn);
  while (blockIt.hasNext()) {
    // Process the blocks for the node until we find one that needs
    // replication
    if (blocksProcessed >= blocksPerLock) {
      blocksProcessed = 0;
      namesystem.writeUnlock();
      namesystem.writeLock();
    }
    blocksProcessed++;
    if (nextBlockAddedToPending(blockIt, dn)) {
      // Exit the inner "block" loop so an iterator for the next datanode
      // is used for the next block.
      pendingCount++;
      break;
    }
  }

It releases the write lock and immediately reacquires it. The blocksProcessed variable tracks how many blocks have been processed. Once it reaches the blocksPerLock threshold, releasing and reacquiring the write lock gives other threads waiting on the write lock a chance to run. This avoids holding the write lock for long periods and blocking other threads, improving concurrency.

💡

Why does check call processPendingReplication first and then moveBlocksToPending? Can it just call one?

In some cases only one function is enough to handle a block’s replication state, but in other cases both are needed. Here is why both calls are necessary in some scenarios:

1. processPendingReplication:
   • Its main job is to handle blocks currently in the “pending replication” state and ensure they reach the required replica count. It is usually invoked periodically by the scheduler to keep cluster blocks sufficiently redundant.
   • When called, it checks which blocks are currently pending and tries to start replication tasks for them.
2. moveBlocksToPending:
   • Its main job is to move blocks into the “pending replication” queue under specific conditions, for example when a node enters maintenance mode or goes offline, so they can be picked up by subsequent replication tasks.
   • When called, it checks whether any node is currently in maintenance or offline, and then adds the relevant blocks to the pending-replication queue.

Why call both at the same time? Because they do not address the same problem:

• processPendingReplication focuses on blocks that are already pending, while moveBlocksToPending focuses on the blocks that need to be handled after a node’s state changes.
• At any moment new blocks may need replicating, or blocks may need to be moved into the pending queue because a node’s state changed. Calling both ensures replication-state handling is comprehensive and timely.

1. checkForCompletedNodes iterates over node states and, if a node’s block migration is complete, adds the node to the toRemove list.
2. processCompletedNodes takes the toRemove list, iterates through it, and sets final state (for example decommissioned), removing the node from outOfServiceNodeBlocks and pendingRep.