Hadoop3.2.1 【 HDFS 】源码分析 : block放置策略

心已赠人 2022-11-21 11:54 159阅读 0赞

### List of articles ###

*  一.前言
 *  二.流程概览.
 *  三. getAdditionalBlock()方法的逻辑流程图。
 *  四.分配数据节点——chooseTarget4NewBlock()

# 一.前言 #

本问主要说client上传文件的时候,向namenode申请新的block的流程和 NameNode如何为block分配存储在哪个DateNode之上.

# 二.流程概览. #

DFSClient在写数据时， 首先会调用ClientProtocol.create()方法在Namenode的文件系统目录树中创建一个HDFS文件， 这个请求在Namenode侧会由FSNamesystem.startFile()方法响应。 startFile()方法会创建INode对象并将INode对象加入FSDirectory中， 然后添加租约，并将操作写入editlog中。成功创建INode对象后， DFSClient会调用ClientProtocol.addBlock()请求分配新的数据块， Namenode会调 FSNamesystem.getAdditionalBlock()方法响应分配请求并返回新申请的数据块， 以及存储这个数据块副本的Datanode信息。

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3poYW5nbG9uZ180NDQ0_size_16_color_FFFFFF_t_70_pic_center]

# 三. getAdditionalBlock()方法的逻辑流程图。 #

■ 首先调用validateAddBlock()方法检查Namenode是否有写权限， 是否处于安全模式中， 以及当前文件系统中的对象是否过多。 然后对当前Namenode中记录的该文件的最后一个数据块与Client上报的最后一个数据块进行比较， 最后判断是否发生了RPC请求的重传， 或者请求异常等情况。  
■ 对于Namenode已经完成了数据块分配操作， 但是Client未能正确收到响应而重发请求的情况， 直接返回Namenode中保存的上一次成功分配的数据块。  
■ 在其他情况下， 首先调用chooseTargetForNewBlock()为新分配的数据块选择保存副本的Datanode。  
■ 再次调用analyzeFileState()方法以防止chooseTarget4NewBlock()方法执行期间Namenode状态发生改变， 例如在调用FSDirWriteFileOp.chooseTarget4NewBlock()方法时， 上一次触发的分配请求执行完成， 这时无须再次分配新的数据块， 返回上一次分配的数据块即可。  
■ 由于Client调用ClientProtocol.addBlock()请求时， 还会顺带提交文件的最后一个数据块， 所以getAdditionalBlock()会调用commitOrCompleteLastBlock()方法提交文件的最后一个数据块， 如果这个数据块满足最小的副本系数， 则将这个数据块的状态转换为COMPLETE。  
■ 最终情况是， 创建新的数据块并保存在INode对象中， 之后将新创建的数据块返回至Client。

/** * * getAdditionalBlock()方法首先会检查文件系统状态， * 然后为新添加的数据块选择存放副本的Datanode， * * 最后构造Block对象并调用FSDirectory.addBlock()方法将Block对象加入 * 文件对应的INode对象中。 * * * The client would like to obtain an additional block for the indicated * filename (which is being written-to). Return an array that consists * of the block, plus a set of machines. The first on this list should * be where the client writes data. Subsequent items in the list must * be provided in the connection to the first datanode. * * Make sure the previous blocks have been reported by datanodes and * are replicated. Will return an empty 2-elt array if we want the * client to "try again later". */
      LocatedBlock getAdditionalBlock(
          String src, long fileId, String clientName, ExtendedBlock previous,
          DatanodeInfo[] excludedNodes, String[] favoredNodes,
          EnumSet<AddBlockFlag> flags) throws IOException { 
    
    
        final String operationName = "getAdditionalBlock";
        NameNode.stateChangeLog.debug("BLOCK* getAdditionalBlock: {} inodeId {}" +
            " for {}", src, fileId, clientName);
    
        LocatedBlock[] onRetryBlock = new LocatedBlock[1];
        FSDirWriteFileOp.ValidateAddBlockResult r;
        checkOperation(OperationCategory.READ);
        final FSPermissionChecker pc = getPermissionChecker();
        readLock();
        try { 
          checkOperation(OperationCategory.READ);
    
          // 验证权限
          r = FSDirWriteFileOp.validateAddBlock(this, pc, src, fileId, clientName,
                                                previous, onRetryBlock);
        } finally { 
          readUnlock(operationName);
        }
    
        if (r == null) { 
          assert onRetryBlock[0] != null : "Retry block is null";
          // This is a retry. Just return the last block.
          return onRetryBlock[0];
        }
        //选取存储datanode节点
        DatanodeStorageInfo[] targets = FSDirWriteFileOp.chooseTargetForNewBlock(
            blockManager, src, excludedNodes, favoredNodes, flags, r);
    
        checkOperation(OperationCategory.WRITE);
        writeLock();
        LocatedBlock lb;
        try { 
          checkOperation(OperationCategory.WRITE);
    
          //保存block
          lb = FSDirWriteFileOp.storeAllocatedBlock(
              this, src, fileId, clientName, previous, targets);
    
        } finally { 
          writeUnlock(operationName);
        }
        getEditLog().logSync();
        return lb;
      }

# 四.分配数据节点——chooseTarget4NewBlock() #

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3poYW5nbG9uZ180NDQ0_size_16_color_FFFFFF_t_70_pic_center 1]

FSNamesystem.getAdditionalBlock()方法在分配数据块时， 会调用FSDirWriteFileOp.chooseTargetForNewBlock => BlockManager.chooseTarget4NewBlock() => BlockPlacementPolicyDefault.chooseTarget()方法为指定数据块选择保存这个数据块副本的Datanode。

chooseTarget()的逻辑实现:

\*\*  
■ 1.客户端在请求NameNode的时候,有一个参数AddBlockFlag,这个参数用于控制要不要在client端写入block.默认值为null, 如果为null的话,会在client写入blcok. 比如:如果在datanode节点上传文件, client端是datanode, 优先会上传到这个datanode上面,否则的话随机.  
■ 2.如果Client不是一个DataNode， 则在集群范围内随机选择一个节点作为第一个节点 。 验证副本数是否足够,足够直接返回,不够继续向下执行.比如: 如果使用java客户端,远程上传文件,client不是datanode , 那么文件会随机挑选一个datanode节点上传文件.  
■ 3.如果Client是一个DataNode， 则判断Client所在的本地数据节点是否符合存储数据块的要求， 如果符合， 则第一个节点分配完毕； 如果该数据节点不符目标节点要求， 则在Client同一个机架范围内寻找， 如果找到目标节点， 则第一个节点分配完毕； 如果在同一个机架内未找到符合要求的目标节点， 则在集群内随机分配一个节点， 找到则第一个节点分配完毕； 否则分配失败。验证副本数是否足够,足够直接返回,不够继续向下执行.  
■4. 如果已经成功分配第一个数据节点， 则在与第一个分配节点不同机架的远程机架内寻找第二个目标节点。 如果符合要求， 则第二个节点分配完毕； 如果在远程机架内未找到符合要求的目标节点， 则在第一个分配节点的机架内寻找， 如果找到则第二个节点分配完毕； 否则第二个节点分配失败。验证副本数是否足够,足够直接返回,不够继续向下执行.  
■ 5.如果前两个节点分配成功， 则准备分配第三个副本的目标节点。 首先判断前两个节点是否在同一个机架内， 如果是， 则在远程机架内寻找目标节点， 找到则第三个节点分配完毕； 如果前两个节点在不同的机架内， 且当前数据块为新分配的数据块， 则在与Client相同的机架内寻找。 如果当前数据块为已有的数据块， 则在第二个节点的机架内分配。 如果找到则第三个节点分配完毕， 未找到则在集群中随机分配一个节点； 否则第三个节点分配失败。验证副本数是否足够,足够直接返回,不够继续向下执行.  
■ 6.如果需要分配的节点数目大于三个， 则在集群范围内随机寻找节点。验证副本数是否足够,足够直接返回,不够继续向下执行.\*\*

protected Node chooseTargetInOrder(int numOfReplicas, 
                                     Node writer,
                                     final Set<Node> excludedNodes,
                                     final long blocksize,
                                     final int maxNodesPerRack,
                                     final List<DatanodeStorageInfo> results,
                                     final boolean avoidStaleNodes,
                                     final boolean newBlock,
                                     EnumMap<StorageType, Integer> storageTypes)
                                     throws NotEnoughReplicasException { 
        final int numOfResults = results.size();
        if (numOfResults == 0) { 
          // 第一个副本
          // 选取本地节点存储 , 如果本地存储节点不可用, 相同机架中随机一台
          DatanodeStorageInfo storageInfo = chooseLocalStorage(writer,
              excludedNodes, blocksize, maxNodesPerRack, results, avoidStaleNodes,
              storageTypes, true);
    
          writer = (storageInfo != null) ? storageInfo.getDatanodeDescriptor()
                                         : null;
    
          if (--numOfReplicas == 0) { 
            return writer;
          }
        }
    
        // 第一台 副本所在的机器
        final DatanodeDescriptor dn0 = results.get(0).getDatanodeDescriptor();
        if (numOfResults <= 1) { 
          /** * * 选取第二个副本 : * * 从远程机架节点选取, 如果节点数量不足,则随机选取 * */
          chooseRemoteRack(1, dn0, excludedNodes, blocksize, maxNodesPerRack,
              results, avoidStaleNodes, storageTypes);
          if (--numOfReplicas == 0) { 
            return writer;
          }
        }
        if (numOfResults <= 2) { 
          //选取第三个节点
    
          // 获取第二个副本所在的节点信息
          final DatanodeDescriptor dn1 = results.get(1).getDatanodeDescriptor();
    
          //如果 第一个机器和第二个机器在同一机架, 换一个机架选取
          //如果远程机架节点不足, 这从第一台机器所在的节点随机选一台.
          if (clusterMap.isOnSameRack(dn0, dn1)) { 
            chooseRemoteRack(1, dn0, excludedNodes, blocksize, maxNodesPerRack,
                results, avoidStaleNodes, storageTypes);
          } else if (newBlock){ 
            //如果是一个新的newBlock , 从第二个节点所在的机架随机选一台
            chooseLocalRack(dn1, excludedNodes, blocksize, maxNodesPerRack,
                results, avoidStaleNodes, storageTypes);
          } else { 
            // 本地机架随机一条
            chooseLocalRack(writer, excludedNodes, blocksize, maxNodesPerRack,
                results, avoidStaleNodes, storageTypes);
          }
          if (--numOfReplicas == 0) { 
            return writer;
          }
        }
        ///三个以上的副本, 随机选取一台....
        chooseRandom(numOfReplicas, NodeBase.ROOT, excludedNodes, blocksize,
            maxNodesPerRack, results, avoidStaleNodes, storageTypes);
        return writer;
      }

参考:  
Hadoop 2.X HDFS源码剖析 – 徐鹏

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3poYW5nbG9uZ180NDQ0_size_16_color_FFFFFF_t_70_pic_center]: /images/20221120/4602cd96394c41849263b581978be00b.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3poYW5nbG9uZ180NDQ0_size_16_color_FFFFFF_t_70_pic_center 1]: /images/20221120/1963a54033b347d0a8f0bd3aec006088.png