HDFS Decommission 代码解读与分析

HDFS 节点退役背景

目前有 200+ 台、1.22PB 的 HDFS3 集群，以及 100- 台、480T 的 HDFS 集群，节点退役（decommission）速度较慢，每小时只能迁移大约 150k 个 block，平均每台节点有约 600k 个 block，单台退役需要约 4 小时。

节点掉线与节点退役的区别

节点退役（Node Decommissioning）

首先，正常情况下将节点加入退役名单，先告知 NameNode 和 YARN 不要再向其提交新任务以及写入数据；然后等待该节点上的数据块在集群中完成复制；此时被退役的节点会优先作为 srcNode 数据源（优先选择处于退役中的节点作为复制数据源 src，因为它没有写入请求，负载较低），其他节点从这个退役中的节点作为复制数据源 src。其他节点从该退役节点复制数据块到其他节点，因此该节点此时的负载会比较高。所有数据块复制完毕后，节点状态变为 Decommissioned，可以从 NameNode URL 页面查看。注意，DataNode 开始复制退役数据的时间同样是 10 分 30 秒之后，并不会因为是主动退役而提前开始，因为 NameNode 与 DataNode 之间始终维持着简单的主从关系，NameNode 不会主动向 DataNode 发起任何 IPC 调用，DataNode 需要配合 NameNode 完成的所有操作都是通过两者心跳应答时携带的 DatanodeCommand 返回的。

节点掉线（Node Drop）

例如强制停止 DataNode、物理机挂了（比如高负载掉线、突发网络故障、硬件故障等），这些都属于节点掉线，一般默认 10 分 30 秒后（主要由两个参数控制）NameNode 会检测到该节点通信异常、判定其掉线。随后 NameNode 根据该节点的 IP，查出该节点上所有的 blockid，以及对应副本所在的机器，通过心跳机制安排数据复制。此时进行数据复制时，数据源不再是已掉线的节点，而是多个副本中的其他节点；同样，此时的副本复制也遵循机架感知、副本搁置策略。

💡 节点掉线与退役两者的区别不仅体现在数据复制方式不同，还在于 NameNode 针对 Under-Replicated Blocks 的数据复制策略也不一样（block 的复制等级分为 5 种）。举个极端的例子：单副本节点退役数据不会丢失，而单副本节点掉线则会真正导致数据丢失。

节点掉线引发的网络风暴

几十 T、甚至上百 T、上百万 block 的节点掉线时，会出现大量 RPC 风暴，尤其对大规模、高负载集群来说，对 NameNode 是很大的挑战，不仅会影响生产性能，还会存在很大的隐患，特别是对于带宽有瓶颈限制的集群。

一般来说，NameNode 检测 DataNode 是否掉线的时间是 103s（心跳时间）+ 25min（NameNode 检测时间，参数为：dfs.namenode.heartbeat.recheck-interval）= 10 分 30 秒。如果在 10min30s 的时间内带宽持续打满，RPC 请求延迟，DataNode 与 NameNode 节点通信不畅，就非常容易导致其他节点继续掉线，形成恶性循环，这种情况应当如何避免？

NameNode 维护着一个复制优先级队列，对于副本不足的文件 block，按优先级排序，仅剩一个副本的文件 block 会享有最高的复制优先级。因此从这里看，对于两副本的集群来说，只要有一个 block 出现异常，就只剩一个副本，就是最高优先级的块复制，会进入风暴模式复制，如果控制不好很容易影响集群性能，甚至搞挂集群。因此一般不建议集群副本因子为 2。

💡 L1（最高）：存在数据丢失风险的块，例如：1. 只有一个副本的块（尤其对于 2 副本的块，下线一台节点）或者这些 block 当前有 0 个 active 副本；2. 单副本位于正在退役节点上的块。

L2：block 副本的实际值远低于配置值（例如 3 副本缺失 2 个），即副本数不到期望值 1/3 时的块，这些 block 会以第二优先级进行复制。例如 4 副本的 block 丢失或损坏了 3 个，它的复制优先级会高于 4 副本缺失 2 个的 block。

L3：副本不足但优先级没有 L2 高的那些块，会被优先复制。第三优先级。

L4：block 已满足最小副本数要求。副本度需求低于 L2-L3。

L5：已损坏的块，但当前仍存在可用的非损坏副本。

通过参数控制节点掉线引发的 RPC 风暴

下述三个参数都是 hdfs-site.xml 中的参数，具体可以参考 Apache Hadoop 官网。其实 block 的复制速度由两方面决定：一是 NameNode 分发任务的速度，二是 DataNode 之间进行复制的速度。前者可以理解为入口，后者可以视为出口。

这些参数在 Hadoop 2.7.3 中并没有。

入口参数

从 NameNode 层面控制任务分发，此参数修改必须重启 NameNode，无需重启 DataNode。

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

该参数决定了当 NameNode 与 DataNode 心跳（3s）下发任务列表时，告诉每个 DataNode 可以进行复制的 block 数量。例如集群有 500 个节点，此值设置为 10，那么一次心跳 NameNode 可下发 DataNode 进行复制的数据块数量为 10*500=5000 块。

假如一个节点掉线或退役时有 800000 个 block 需要复制，那么 NameNode 需要多长时间才能将待复制 block 的任务全部下发给 DataNode？

任务分发时间 = 
待复制 block 数 /（DataNode 数 * 参数值）* 心跳间隔时间

出口参数

与上面从 NameNode 层面控制任务分发相对应，下面两个参数从 DataNode 层面控制，同样需要重启 NameNode。

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

此参数用于控制 DataNode 节点进行数据复制的最大线程数。从上文我们知道，block 的复制优先级分为 5 种。此参数控制的是除最高优先级以外的块复制，也就是除最高优先级以外的复制流限制。

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

此参数用于控制 DataNode 所有优先级块复制的流数量，包含最高优先级；一般与上面两个参数互相配合使用。

💡 前者参数控制 DataNode 接受任务的频率，后两个参数进一步限制 DataNode 一次可完成的最大并行线程网络传输量。以上参数的具体取值，取决于集群规模和集群配置，不能一概而论。一般来说从入口控制会比较简单。例如 500 台规模的集群，dfs.namenode.replication.work.multiplier.per.iteration = 10，那么集群一次心跳下发 5000 个 block 的量；假设集群文件存储完全打散到 500 台节点，每个节点同时复制 10 个 block（实际由于副本搁置策略、机架感知等原因，并不会所有节点都参与数据复制），每个 block 大小 128Mb，那么每个节点的网络负载是 128 * 10/3 = 546Mb/s，这时就要结合实际查看是否存在带宽瓶颈，这么大的网络 IO 是否会影响正常任务的计算，如果有的话，这个值就要适当调小一些。

如何快速下线

让节点快速下线的本质其实就是提升副本的复制速度。主要还是依靠上面三个参数进行控制：第一是控制 NameNode 任务分发，第二是控制 DataNode 复制速率，前提是不影响正常生产任务的进行。集群规模越小，下线越慢，例如因为下发的总量会慢很多。

比如一个 500 台节点的集群 dfs.namenode.replication.work.multiplier.per.iteration=10，而一个 50 台节点的集群这个值就应设为 100，两者 NameNode 任务分发速度才能保持一致。具体需要结合实际集群规模进行设置。

<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>

在测试集群中，15 台 HDFS 节点，平均每台 80k block，每小时能迁移 63k block，比最初的 12k 提升了 5 倍。

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

影响

会显著提升 CPU Load、CPU IO Wait、Disk IO Wait 等资源使用率。

源码分析

当 dfs.hosts.exclude 的地址写入 conf 之后，refreshNodes 就可以拿到 exclude list 并对其进行操作。

/**
 * 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();
  }
}

通过 dfs.exclude 读取节点地址，然后将其标记为 decommission，并使用 monitor 来追踪该节点。

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());
      monitor.startTrackingNode(node);
    }
  } else {
    LOG.trace("startDecommission: Node {} in {}, nothing to do.",
        node, node.getAdminState());
  }
}

在 decommission 之前，会先对集群的 stats 进行预减，例如 capacityUsed、capacityTotal 等。

monitor 的 run() 方法会调用 processPendingNodes()，从 pending nodes 中取出节点并进行 decommission。

/**
 * 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);
  }
}

拿到 nodes 之后，会调用 check() 方法，获取这些节点的 block 信息并处理 PendingReplication 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() 方法会调用 isBlockReplicatedOk() 方法（核心），用于判断 block 是否需要 reconstruction，并将相关信息放入 blockManager 的 *replication queue* 中。

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

2. moveBlocksToPending() 方法会判断队列的长度以及 pending replication 的上限，然后为每一个需要 maintenance 或 decommission 的节点创建一个 block 迭代器，随后每个 block 都会继续调用上面提到的 isBlockReplicatedOk() 方法，直到达到 block 复制上限才停止并退出循环。

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;
    }
  }

它在释放写锁之后又立刻重新申请写锁，这是因为 blocksProcessed 变量用于追踪已处理的 block 数量。当已处理的 block 数达到预定阈值 blocksPerLock 时，通过释放当前写锁并再次获取写锁，可以让其他等待写锁的线程有机会执行。这有助于避免因长时间持有写锁而阻塞其他线程，从而提升并发性能。

💡

为什么 check 函数要先调用 processPendingReplication，然后再调用 moveBlocksToPending？不能只调用其中一个吗？

在某些情况下，只调用其中一个函数就足以处理 block 的复制状态，但在另一些情况下，可能需要同时调用这两个函数。让我们看看为什么在某些情况下需要同时调用这两个函数：

1. **processPendingReplication**：
   • 这个函数的主要任务是处理当前处于 “pending replication” 状态的 block，确保它们的副本数达到所需数量。它通常会被调度程序周期性地调用，以确保集群中的 block 保持足够的冗余度。
   • 调用该函数时，会检查当前有哪些 block 处于 pending 状态，并尝试启动复制任务来复制这些 block。
2. **moveBlocksToPending**：
   • 这个函数的主要任务是在特定情况下（例如某个节点进入 maintenance 模式或下线时），将 block 移动到 “pending replication” 队列中，等待后续的复制任务处理。
   • 调用该函数时，会检查当前是否有节点处于 maintenance 模式或下线状态，然后将相关 block 添加到待复制队列中。

为什么需要同时调用这两个函数呢？因为它们处理的问题并不完全相同：

• processPendingReplication 主要关注当前正在 pending 的 block，而 moveBlocksToPending 主要关注节点状态发生变化后需要处理的 block。
• 在某些情况下，可能会有新的 block 需要复制，或由于节点状态变化而需要将部分 block 移入等待复制的队列，因此同时调用这两个函数可以保证 block 复制状态处理的全面性和及时性。

1. checkForCompletedNodes 方法会遍历检查节点状态，如果 block 的迁移任务已完成，就将该节点添加到 toRemove 列表中。
2. processCompletedNodes 方法拿到 toRemove 列表，遍历后设置状态，例如改为 decommissioned，并将其从 outOfServiceNodeBlocks 和 pendingRep 中移除。