HDFS集群的启动(2)——DataNode节点的注册

深藏阁楼爱情的钟 2022-06-18 11:16 427阅读 0赞

前面我已经就HDFS集群的启动问题在整体上进行了阐述，而在本文，我将主要针对DataNode节点在启动的过程中会首先向NameNode节点进行注册这一细节进行深入的讨论。

先来简单的讲一讲DataNode节点向NameNode节点进行注册的目的吧！DataNode节点向NameNode节点注册，一是告诉NameNode节点自己提供服务的网络地址端口，二是获取NameNode节点对自己的管理与控制。我们说，在一个HDFS集群中有一个NameNode节点和若干DataNode节点(上千台机器，每个机器至少一个节点)。我们不可能给每一个客户端关于所有该集群中的DataNode节点的服务地址，而是让NameNode节点来记住所有的DataNode节点信息，然后客户端通过NameNode节点来获取DataNode节点的信息。当然，真实的情况是，由NameNode节点来收集所有的DataNode的当前工作情况，并以此来负责整个集群的负载均衡与任务的合理安排/调度，即告诉客户端它应该调用那些DataNode节点来获取服务。好吧，现在我们来正式的看看DataNode节点的注册过程吧，实际上主要是NameNode节点是如何处理的？

DataNode节点调用DatanodeProtocol协议(远程调用接口)来向NameNode节点进行注册，那么NameNode节点是如何处理的呢？当然，在这里我不得不提的是，DataNode向NameNode发送了那些注册信息呢？这些信息包括：![0_1320325346k64u.gif][]

其中，name表示DataNode节点提供数据传输/交换的服务地址端口，storageID表示DataNode节点的存储器在整个HDFS集群中的全局编号(当然，这个编号也就由NameNode统一分配)，infoPort表示查询DataNode节点当前状态信息的端口号，ipcPort表示DataNode节点提供ClientDatanodeProtocol服务端口号。接下来还是重点看看NameNode节点的处理吧！

当NameNode节点收到某一个DataNode节点的注册请求之后，它会首先检查DataNode节点所运行的HDFS系统是否和自己是同一个版本号，然后才交给FSNamesystem来处理，源代码：

![0_1320326306p21g.gif][]

在FSNamesystem中，会首先判断该DataNode是否被允许被连接到NameNode节点，然而，这是如何进行判断的呢？在配置文件中，有这样两个选项：dfs.hosts和dfs.hosts.exclude，它们的值对应的都是一个文件路径，dfs.hosts指向的文件是一个允许连接到NameNode节点的主机列表，dfs.hosts.exclude指向的文件是一个不允许连接到NameNode节点的主机列表，所以在NameNode节点创建FSNamesystem的时候，会根据配置文件中的选项值从文件中加载这些主机列表，就这么简单。如果这时发现DataNode的storageID为空，会立马给它分配一个全局ID，然后，FSNamesystem会根据DataNode的ip地址把它映射到合适的机架(rack)中，哦，简单的提一下这样做的目的(这个问题我以后会讲到)：在大多数情况下，副本系数是3，HDFS的存放策略是将一个副本存放在本地机架的节点上，一个副本存放在同一个机架的另一个节点上，最后一个副本放在不同机架的节点上。这种策略减少了机架间的数据传输，这就提高了数据块写操作的效率。机架的错误远远比节点的错误率低，所以这个策略不会影响到数据的可靠性和可用性。与此同时，因为数据块只放在了两个(不是三个)机架上，所以此策略减少了读取数据块是需要的网络传输总带宽。在这种策略下，副本并不是均匀分布在不同的机架上。三分之二的副本在同一个机架上，其它副本均匀分布在剩下的机架上。这一策略在不损害数据可靠性和读取性能的情况下，尽可能的调高了写的效率。关于机架的概念本文就先谈到这儿，回过头来再看看NameNode接下来的处理吧！另外FSNamesystem还会对该DataNode的描述信息和它对应的storageID做一个映射并保存起来，之后，就把该DataNode节点加入到heartbeats集合中，让HeartbeartMinitor后台线程来实时监测该DataNode节点当前是否还alive。整个注册过程到此也就基本结束了。当然，NameNode节点会把注册结果信息返回给DataNode节点。

还是整一段源代码先：

public synchronized void registerDatanode(DatanodeRegistration nodeReg) throws IOException {
        String dnAddress = Server.getRemoteAddress();
        if (dnAddress == null) {
          // Mostly called inside an RPC.
          // But if not, use address passed by the data-node.
          dnAddress = nodeReg.getHost();
        }      
    
        // check if the datanode is allowed to be connect to the namenode
        if (!verifyNodeRegistration(nodeReg, dnAddress)) {
          throw new DisallowedDatanodeException(nodeReg);
        }
    
        String hostName = nodeReg.getHost();
          
        // update the datanode's name with ip:port
        DatanodeID dnReg = new DatanodeID(dnAddress + ":" + nodeReg.getPort(),
                                          nodeReg.getStorageID(),
                                          nodeReg.getInfoPort(),
                                          nodeReg.getIpcPort());
        nodeReg.updateRegInfo(dnReg);
          
        NameNode.stateChangeLog.info(
                                     "BLOCK* NameSystem.registerDatanode: "
                                     + "node registration from " + nodeReg.getName()
                                     + " storage " + nodeReg.getStorageID());
    
        DatanodeDescriptor nodeS = datanodeMap.get(nodeReg.getStorageID());
        DatanodeDescriptor nodeN = host2DataNodeMap.getDatanodeByName(nodeReg.getName());
          
        if (nodeN != null && nodeN != nodeS) {
          NameNode.LOG.info("BLOCK* NameSystem.registerDatanode: " + "node from name: " + nodeN.getName());
          // nodeN previously served a different data storage,
          // which is not served by anybody anymore.
          removeDatanode(nodeN);
          // physically remove node from datanodeMap
          wipeDatanode(nodeN);
          nodeN = null;
        }
    
        if (nodeS != null) {
          if (nodeN == nodeS) {
            // The same datanode has been just restarted to serve the same data
            // storage. We do not need to remove old data blocks, the delta will
            // be calculated on the next block report from the datanode
            NameNode.stateChangeLog.debug("BLOCK* NameSystem.registerDatanode: " + "node restarted.");
          } else {
            // nodeS is found
            /* The registering datanode is a replacement node for the existing
              data storage, which from now on will be served by a new node.
              If this message repeats, both nodes might have same storageID
              by (insanely rare) random chance. User needs to restart one of the
              nodes with its data cleared (or user can just remove the StorageID
              value in "VERSION" file under the data directory of the datanode,
              but this is might not work if VERSION file format has changed
           */        
            NameNode.stateChangeLog.info( "BLOCK* NameSystem.registerDatanode: " + "node " + nodeS.getName() + " is replaced by " + nodeReg.getName() + " with the same storageID " + nodeReg.getStorageID());
          }
          // update cluster map
          clusterMap.remove(nodeS);
          nodeS.updateRegInfo(nodeReg);
          nodeS.setHostName(hostName);
          
          // resolve network location
          resolveNetworkLocation(nodeS);
          clusterMap.add(nodeS);
            
          // also treat the registration message as a heartbeat
          synchronized(heartbeats) {
            if( !heartbeats.contains(nodeS)) {
              heartbeats.add(nodeS);
              //update its timestamp
              nodeS.updateHeartbeat(0L, 0L, 0L, 0);
              nodeS.isAlive = true;
            }
          }
          
          return;
        }
    
        // this is a new datanode serving a new data storage
        if (nodeReg.getStorageID().equals("")) {
          // this data storage has never been registered
          // it is either empty or was created by pre-storageID version of DFS
          nodeReg.storageID = newStorageID();
          NameNode.stateChangeLog.debug( "BLOCK* NameSystem.registerDatanode: " + "new storageID " + nodeReg.getStorageID() + " assigned.");
        }
        
        // register new datanode
        DatanodeDescriptor nodeDescr  = new DatanodeDescriptor(nodeReg, NetworkTopology.DEFAULT_RACK, hostName);
        
        LOG.debug("start to resolve the network location of DataNode["+nodeReg.name+"].");
        resolveNetworkLocation(nodeDescr);
        
        unprotectedAddDatanode(nodeDescr);
        clusterMap.add(nodeDescr);
          
        // also treat the registration message as a heartbeat
        synchronized(heartbeats) {
          heartbeats.add(nodeDescr);
          nodeDescr.isAlive = true;
          // no need to update its timestamp
          // because its is done when the descriptor is created
        }
        
        return;
      }

这里，我想提一下上面代码中蓝颜色的部分，这一部分有什么作用？我们知道在DataNoe节点注册成功之后，就会开始不断的向NameNode节点发送心跳包，来告诉它自己当前工作是正常的。在NameNode节点有一个后台工作线程会定期的检测哪儿数据节点没有按时发送心跳包，对于那些没有按时发送心跳包的数据节点就认为它们是不可用的，并清除与它们相关的信息(最重要的当然还是存储在它上面的Block信息)。另外，一台机器一般对应一个数据节点，当也有可能对应多个数据节点，一个数据节点唯一对应一个存储器(可以作是HDFS的一个逻辑磁盘)。但是当一个DataNode节点注册的时候，会有一些特殊的情况，对于这些特殊的情况也有特殊的处理以此来提高系统性能(就是蓝颜色代码额外干的事)。在介绍这些特殊处理之前，我想先介绍一下与之有关的两个数据结构：

NavigableMap<String, DatanodeDescriptor> datanodeMap = new TreeMap<String, DatanodeDescriptor>();
    
    Host2NodesMap host2DataNodeMap = new Host2NodesMap();

datanodeMap保存的是 注册的storageID与它所在的DataNode节点之间的映射，host2DataNodeMap保存的是注册的host与它上面的DataNode节点之间的映射，请注意这里的host不仅仅指的是一个ip地址，还包含一个端口号，说白了host就是该DataNode节点的数据服务地址。关于注册时的特殊情况及其处理如下：

1.一台DataNode节点突然宕机了，然后立马恢复重启，NameNode节点也没有及时检测到，那么当这台重启的DataNode节点注册时，就会出现nodeN == nodeS!= null，在这种情况下NameNode节点不会清除与该节点相关的信息，而只是会更新该节点的状态信息，同时重新解析它的ip地址到HDFS集群的网络拓扑图中(这里之所以会重新解析ip地址，是因为在一个DataNode节点重启的过程中唯一可变的就是它所在的网络拓扑结构发生了变化)；

2.一台DataNode节点在突然宕机或者认为stop之后，又马上重启，但是此时该DataNode管理的存储器(逻辑磁盘)并不是之前管理的那一个存储器了，那么该节点注册时就会出现nodeN != null 、nodeS !=nodeN ，对于这种情况就相当于是该DataNode节点第一次注册，但是又必须要清除该DataNode节点上一次的注册信息，特别是Block与DataNode之间的映射信息；

3.当一个DataNode节点由于更换了ip地址，所以它必须要重启并重新向NameNode节点注册，此时该节点注册时就会出现nodeS != null、nodeS != nodeN(存储器没有变)，此时并不需要清除该存储器原来所在的DataNode节点的相关注册信息，主要是Block和DataNode节点之间的映射信息，而只需要更新原来与存储器绑定的DataNode节点的服务器地址信息，当然此时还必须要把新的DataNode节点的ip地址解析到HDFS集群的网络拓扑图中。

[0_1320325346k64u.gif]: /images/20220618/e8b989b144a845ac842c5d9cffc1de01.png
[0_1320326306p21g.gif]: /images/20220618/5aef7506154c42a4a2f5a720328c7e9e.png

发表评论取消回复

表情：

评论列表（有 0 条评论，427人围观）

还没有评论，来说两句吧...

相关阅读

相关 hadoop 集群启动从节点无datanode

问题 hadoop 集群启动从节点无datanode 具体问题笔者第一次启动hadoop 集群时，集群所需线程全部正常启动，时隔几天，再次启动，从节点无

- 日理万妓/ 2023年09月24日 12:51/ 0 赞/ 145 阅读

相关 hadoop集群启动后datanode没有启动

今天启动[hadoop][]集群后，查看进程发现名称节点的NameNode进程启动了，但是数据节点的DataNode进程没有启动，如下：这个问题是个常见问题，对于自学的初学

痛定思痛。/ 2023年09月23日 15:50/ 0 赞/ 169 阅读

相关 HDFS核心源码解析(二)——DataNode的启动及注册流程

申明：本文基于hadoop2.7 进行源码研读一、DataNode类代码注释我简单对类注释做了一些翻译： / DataNode is a class (

骑猪看日落/ 2022年09月09日 04:57/ 0 赞/ 269 阅读

相关 HDFS集群的启动(2)——DataNode节点的注册

前面我已经就HDFS集群的启动问题在整体上进行了阐述，而在本文，我将主要针对DataNode节点在启动的过程中会首先向NameNode节点进行注册这一细节进行深入的讨

深藏阁楼爱情的钟/ 2022年06月18日 11:16/ 0 赞/ 428 阅读

相关 HDFS集群的启动(1)——概述

众所周知，在HDFS集群中，主要有两类节点，即NameNode和DataNode节点，确切的说，一个NameNode节点，其它的所有DataNode节点。那么，H

爱被打了一巴掌/ 2022年06月18日 10:55/ 0 赞/ 416 阅读

相关在启动HDFS时，针对集群中namenode无法识别datanode的问题的解决方法

最近由于重装了系统，需要对之前搭建的集群要做些改动。在对每个虚拟机的网络进行正确的配置之后，重新执行hadoop/sbin/start-dfs.sh命令来启动HDFS，然而na

╰半夏微凉°/ 2022年06月12日 02:36/ 0 赞/ 210 阅读

相关 hadoop集群启动之后dataNode节点没有启动

启动Hadoop集群之后slave机器的dataNode节点没有启动 master机器的nameNode节点启动了 1、在集群/usr/local/src/hadoop/bi

朱雀/ 2022年06月04日 00:36/ 0 赞/ 401 阅读

相关主节点nameNode启动，数据节点dataNode崩溃

原因：是没有执行hdfs nameNode -format 初始化解决：把hadoop\_tmp目录下的文件都删除，执行命令hdfs nameNode -format 初始

男娘i/ 2022年04月05日 13:53/ 0 赞/ 262 阅读

相关 HDFS集群扩容节点

在企业中，经常需要根据业务的发展对HDFS集群进行节点扩容。比如我们要添加一个新DataNode节点到集群中：新节点主机名:bigdata114

ゝ一世哀愁。/ 2021年10月23日 11:46/ 0 赞/ 406 阅读

相关 Hadoop中启动hdfs时DataNode没有启动

是刚开始学习Hadoop，这个问题困扰了我一个多小时，为了以后少犯这种错误，也为了别人能少走歪路，我记录下来我的解决方法，可能不是最快捷的，但是这是我摸索出来的。 hd...

朱雀/ 2020年10月25日 08:10/ 0 赞/ 1017 阅读