HDFS源码分析之UnderReplicatedBlocks(一)

简介:         UnderReplicatedBlocks是HDFS中关于块复制的一个重要数据结构。在HDFS的高性能、高容错性体系中,总有一些原因促使HDFS系统内进行块复制工作,比如基于高性能的负载均衡、基于容错性的数据块副本数恢复等。

        UnderReplicatedBlocks是HDFS中关于块复制的一个重要数据结构。在HDFS的高性能、高容错性体系中,总有一些原因促使HDFS系统内进行块复制工作,比如基于高性能的负载均衡、基于容错性的数据块副本数恢复等。普遍的,任何工作都会有一个优先级的问题,特别是这里的数据块复制,不可能简单的按照先入先出或者其他简单策略,比方说,基于容错性的数据块副本数恢复,特别是数据块副本仅有一个的数据块副本数恢复,其优先级肯定要比基于高性能的负载均衡高,所以数据块复制要有个优先级的概念,那么,数据块复制的优先级怎么确定,怎么存储?一切答案均在UnderReplicatedBlocks中,本文我们将开始分析UnderReplicatedBlocks。

        UnderReplicatedBlocks专门用于存储待优先级的需要复制的数据块。何谓数据块复制优先级,我们看下UnderReplicatedBlocks类中的几个静态成员变量及其说明就会得到答案,如下:

  /** The queue with the highest priority: {@value} */
  // 最高优先级队列的优先级值0
  static final int QUEUE_HIGHEST_PRIORITY = 0;
  /** The queue for blocks that are way below their expected value : {@value} */
  // 第二优先级队列的优先级值1:主要针对低于副本数要求很多的数据块
  static final int QUEUE_VERY_UNDER_REPLICATED = 1;
  /** The queue for "normally" under-replicated blocks: {@value} */
  // 第三优先级队列的优先级值2:主要针对低于副本数要求不是很多,即一般情况的数据块
  static final int QUEUE_UNDER_REPLICATED = 2;
  /** The queue for blocks that have the right number of replicas,
   * but which the block manager felt were badly distributed: {@value}
   */
  // 第四优先级队列的优先级值3:主要针对副本数满足要求,但是数据块管理器BlockManager感觉严重分布不均
  static final int QUEUE_REPLICAS_BADLY_DISTRIBUTED = 3;
  /** The queue for corrupt blocks: {@value} */
  // 第五优先级队列的优先级值4:主要针对损坏的数据块
  static final int QUEUE_WITH_CORRUPT_BLOCKS = 4;
        数据块复制优先级共分为五种级别,从高到底依次如下:

        1、QUEUE_HIGHEST_PRIORITY = 0:最高优先级

        主要针对数据块副本数非常的、严重的不足的情况,当前副本数低于期望值,且仅有1个或者干脆没有,比如副本数仅有1个,或者副本数干脆为0,但是还存在退役副本,这种情况最危险,数据最容易丢失,所以复制的优先级也最高;

        2、QUEUE_VERY_UNDER_REPLICATED = 1:第二优先级

        主要针对数据块副本数比较不足的情况,比上面的情况好点,当前副本数低于期望值,但是副本数大于1,其判断公式为当前副本数curReplicas乘以3还小于期望副本数expectedReplicas,这种情况也比较危险,数据也容易丢失,所以复制的优先级也很高;

        3、QUEUE_UNDER_REPLICATED = 2:第三优先级

        主要针对数据块副本数低于期望值,但是还不是很严重、很危急的情况;

        4、QUEUE_REPLICAS_BADLY_DISTRIBUTED = 3:第四优先级

        主要针对数据块已经有足够的副本数,但是没有足够的机架的情况,这是负载均衡等策略需要的产物;

        5、QUEUE_WITH_CORRUPT_BLOCKS = 4:第五优先级

        主要针对损坏的数据块的情况,其副本数位0,但是还没有退役副本,所以优先级最低,话说,这种数据块还需要哦复制吗?留个小小的疑问吧!

        通过上面的说明,我们可以简单总结下:

        当前副本数低于期望值时,如果当前副本数为1,甚至存在退役副本的情况下为0时,其复制优先级最高,如果当前副本数为0且没有退役副本,则复制优先级最低;如果当前副本数大于1,但是乘以3还小于期望副本数,处于比较危险的情况,则优先级次之,否则是第三优先级。而当当前副本数等于或高于期望值时,则可能是没有足够机架的情况,此时的优先级比最低优先级稍高,为第四优先级。

        UnderReplicatedBlocks也提供了根据数据块及其副本情况来获取复制优先级的getPriority()方法,代码如下:

  /** Return the priority of a block
   * 计算指定数据块复制优先级
   * 
   * @param block a under replicated block
   * @param curReplicas current number of replicas of the block
   * @param expectedReplicas expected number of replicas of the block
   * @return the priority for the blocks, between 0 and ({@link #LEVEL}-1)
   */
  private int getPriority(Block block,
                          int curReplicas, 
                          int decommissionedReplicas,
                          int expectedReplicas) {
    
	// 参数校验:当前副本数curReplicas应大于等于0
	assert curReplicas >= 0 : "Negative replicas!";
	
	// 如果当前副本数curReplicas大于等于期望的副本数,则返回第四优先级队列的优先级值3
    if (curReplicas >= expectedReplicas) {
    	
      // 数据块已经有足够的副本数,但是没有足够的机架
      // Block has enough copies, but not enough racks
      return QUEUE_REPLICAS_BADLY_DISTRIBUTED;
    } else if (curReplicas == 0) {// 如果当前副本数curReplicas为0
    	
      // If there are zero non-decommissioned replicas but there are
      // some decommissioned replicas, then assign them highest priority
      // 如果decommissionedReplicas大于0,返回最高优先级队列的优先级值0
      // 即没有非退役副本,但是有一些退役的副本,那么我们需要分配给它们最高优先级
      if (decommissionedReplicas > 0) {
        return QUEUE_HIGHEST_PRIORITY;
      }
      
      // all we have are corrupt blocks
      // 没有非退役副本,也没有退役副本,我们就认为它是个损坏的数据块,复制优先级最低,为第五优先级队列的优先级值4
      return QUEUE_WITH_CORRUPT_BLOCKS;
    } else if (curReplicas == 1) {// 如果当前副本数curReplicas为1
      //only on replica -risk of loss
      // highest priority
      // 仅仅有一个副本,有丢失的风险,所以赋予最高优先级0
      return QUEUE_HIGHEST_PRIORITY;
    } else if ((curReplicas * 3) < expectedReplicas) {
      //there is less than a third as many blocks as requested;
      //this is considered very under-replicated
      // 如果如果当前副本数curReplicas乘以3还小于期望副本数expectedReplicas,返回第二优先级队列的优先级值1
      return QUEUE_VERY_UNDER_REPLICATED;
    } else {
      //add to the normal queue for under replicated blocks
      // 一般的低于副本数的优先级,返回第三优先级队列的优先级值2
      return QUEUE_UNDER_REPLICATED;
    }
        思路和上述介绍一样,大体逻辑如下:

        1、如果当前副本数curReplicas大于等于期望的副本数,则返回第四优先级队列的优先级值3--QUEUE_REPLICAS_BADLY_DISTRIBUTED;

        2、如果当前副本数curReplicas为0,且如果decommissionedReplicas大于0,返回最高优先级队列的优先级值0,没有非退役副本,也没有退役副本,我们就认为它是个损坏的数据块,复制优先级最低,为第五优先级队列的优先级值4;

        3、如果当前副本数curReplicas为1,仅仅有一个副本,有丢失的风险,所以赋予最高优先级0;

        4、如果如果当前副本数curReplicas乘以3还小于期望副本数expectedReplicas,返回第二优先级队列的优先级值1;

        5、一般的低于副本数的优先级,返回第三优先级队列的优先级值2。


         UnderReplicatedBlocks还提供了涉及复制优先级队列的成员变量,如下:

  /** The total number of queues : {@value} */
  // 队列总数
  static final int LEVEL = 5;
  /** the queues themselves */
  // 队列集合
  private final List<LightWeightLinkedSet<Block>> priorityQueues
      = new ArrayList<LightWeightLinkedSet<Block>>();
  /** Stores the replication index for each priority */
  // 存储每个优先级对应的复制索引
  private Map<Integer, Integer> priorityToReplIdx = new HashMap<Integer, Integer>(LEVEL);
        总的队列数目为5,而存储待复制不同优先级块集合的是priorityQueues列表,它是数据块集合LightWeightLinkedSet的列表,并且还提供了存储每种优先级对应的块复制索引的集合priorityToReplIdx,它是数字形式优先级priority到块在集合LightWeightLinkedSet中位置索引index的映射。

        UnderReplicatedBlocks的构造函数如下:

  /** Create an object. */
  // 构造函数,创建一个对象
  UnderReplicatedBlocks() {
	  
	// 5个LightWeightLinkedSet集合,存储到priorityQueues列表中,
	// 并将优先级与复制索引的映射存储到priorityToReplIdx中
    for (int i = 0; i < LEVEL; i++) {
      priorityQueues.add(new LightWeightLinkedSet<Block>());
      priorityToReplIdx.put(i, 0);
    }
  }
        上来先构造5个LightWeightLinkedSet集合,并按照优先级由高到低的顺序,添加到列表priorityQueues中,并初始化每种块复制优先级对应的位置索引为0。

        UnderReplicatedBlocks还提供了相应的添加、移除数据块及更新优先级方法,分别介绍如下:

        1、添加数据块add()

  /** add a block to a under replication queue according to its priority
   * @param block a under replication block
   * @param curReplicas current number of replicas of the block
   * @param decomissionedReplicas the number of decommissioned replicas
   * @param expectedReplicas expected number of replicas of the block
   * @return true if the block was added to a queue.
   */
  synchronized boolean add(Block block,
                           int curReplicas, 
                           int decomissionedReplicas,
                           int expectedReplicas) {
    assert curReplicas >= 0 : "Negative replicas!";
    
    // 根据入参数据块及其副本情况计算块复制的优先级priLevel
    int priLevel = getPriority(block, curReplicas, decomissionedReplicas,
                               expectedReplicas);
    
    // 如果块复制优先级priLevel小于5(即是一个正确有效的优先级),并且
    // 根据优先级priLevel从priorityQueues中取出相应块集合并将块添加入集合成功的话,返回true
    if(priLevel != LEVEL && priorityQueues.get(priLevel).add(block)) {
      if(NameNode.blockStateChangeLog.isDebugEnabled()) {
        NameNode.blockStateChangeLog.debug(
          "BLOCK* NameSystem.UnderReplicationBlock.add:"
          + block
          + " has only " + curReplicas
          + " replicas and need " + expectedReplicas
          + " replicas so is added to neededReplications"
          + " at priority level " + priLevel);
      }
      return true;
    }
    // 否则返回false
    return false;
  }
        添加数据块的add()方法比较简单,首先根据入参数据块及其副本情况调用getPriority()方法计算块复制的优先级priLevel,然后如果块复制优先级priLevel小于5(即是一个正确有效的优先级),并且根据优先级priLevel从priorityQueues中取出相应块集合并将块添加入集合成功的话,返回true,表示添加成功,否则返回false,表示添加失败。

        2、移除数据块remove()

  /** remove a block from a under replication queue */
  synchronized boolean remove(Block block, 
                              int oldReplicas, 
                              int decommissionedReplicas,
                              int oldExpectedReplicas) {
    
	// 先根据入参数据块及其副本情况,调用getPriority()方法计算块复制优先级priLevel
	int priLevel = getPriority(block, oldReplicas, 
                               decommissionedReplicas,
                               oldExpectedReplicas);
	
	// 调用两个参数的remove()方法,移除数据块
    return remove(block, priLevel);
  }
  /**
   * Remove a block from the under replication queues.
   *
   * The priLevel parameter is a hint of which queue to query
   * first: if negative or >= {@link #LEVEL} this shortcutting
   * is not attmpted.
   *
   * If the block is not found in the nominated queue, an attempt is made to
   * remove it from all queues.
   *
   * <i>Warning:</i> This is not a synchronized method.
   * @param block block to remove
   * @param priLevel expected privilege level
   * @return true if the block was found and removed from one of the priority queues
   */
  boolean remove(Block block, int priLevel) {
    
	// 如果优先级priLevel是正确有效的,且根据优先级priLevel从列表priorityQueues中
	// 取出数据块集合后,从中移除数据块成功的话,返回true,表示移除成功
	if(priLevel >= 0 && priLevel < LEVEL 
        && priorityQueues.get(priLevel).remove(block)) {
      if(NameNode.blockStateChangeLog.isDebugEnabled()) {
        NameNode.blockStateChangeLog.debug(
          "BLOCK* NameSystem.UnderReplicationBlock.remove: "
          + "Removing block " + block
          + " from priority queue "+ priLevel);
      }
      return true;
    } else {
    	
      // 否则,在给定优先级对应数据块集合中移除失败的话,尝试从所有优先级各自对应的队列中移除数据块,
      // 任何一个移除成功,均返回true,表示移除成功
    	
      // Try to remove the block from all queues if the block was
      // not found in the queue for the given priority level.
      for (int i = 0; i < LEVEL; i++) {
        if (priorityQueues.get(i).remove(block)) {
          if(NameNode.blockStateChangeLog.isDebugEnabled()) {
            NameNode.blockStateChangeLog.debug(
              "BLOCK* NameSystem.UnderReplicationBlock.remove: "
              + "Removing block " + block
              + " from priority queue "+ i);
          }
          return true;
        }
      }
    }
	
	// 最后,如果还不行的话,则返回false,表示移除失败
    return false;
  }
        首先,在四个参数的remove()方法中,先根据入参数据块及其副本情况,调用getPriority()方法计算块复制优先级priLevel,然后调用两个参数的remove()方法,移除数据块;

        其次,在两个参数的remove()方法中,如果优先级priLevel是正确有效的,且根据优先级priLevel从列表priorityQueues中取出数据块集合后,从中移除数据块成功的话,返回true,表示移除成功;否则,在给定优先级对应数据块集合中移除失败的话,尝试从所有优先级各自对应的队列中移除数据块,任何一个移除成功,均返回true,表示移除成功;最后,如果还不行的话,则返回false,表示移除失败。

        3、更新优先级update()

  /**
   * Recalculate and potentially update the priority level of a block.
   *
   * If the block priority has changed from before an attempt is made to
   * remove it from the block queue. Regardless of whether or not the block
   * is in the block queue of (recalculate) priority, an attempt is made
   * to add it to that queue. This ensures that the block will be
   * in its expected priority queue (and only that queue) by the end of the
   * method call.
   * @param block a under replicated block
   * @param curReplicas current number of replicas of the block
   * @param decommissionedReplicas  the number of decommissioned replicas
   * @param curExpectedReplicas expected number of replicas of the block
   * @param curReplicasDelta the change in the replicate count from before
   * @param expectedReplicasDelta the change in the expected replica count from before
   */
  synchronized void update(Block block, int curReplicas,
                           int decommissionedReplicas,
                           int curExpectedReplicas,
                           int curReplicasDelta, int expectedReplicasDelta) {
    
	// curReplicas代表当前副本数,curReplicasDelta代表之前发生的副本数变化
	// curExpectedReplicas代表当前期望副本数,expectedReplicasDelta代表之前发生的期望副本数变化
	
	// 计算之前的副本数oldReplicas和之前的期望副本数oldExpectedReplicas
	int oldReplicas = curReplicas-curReplicasDelta;
    int oldExpectedReplicas = curExpectedReplicas-expectedReplicasDelta;
    
    // 计算当前的块复制优先级curPri
    int curPri = getPriority(block, curReplicas, decommissionedReplicas, curExpectedReplicas);
    
    // 计算之前的块复制优先级oldPri
    int oldPri = getPriority(block, oldReplicas, decommissionedReplicas, oldExpectedReplicas);
    if(NameNode.stateChangeLog.isDebugEnabled()) {
      NameNode.stateChangeLog.debug("UnderReplicationBlocks.update " + 
        block +
        " curReplicas " + curReplicas +
        " curExpectedReplicas " + curExpectedReplicas +
        " oldReplicas " + oldReplicas +
        " oldExpectedReplicas  " + oldExpectedReplicas +
        " curPri  " + curPri +
        " oldPri  " + oldPri);
    }
    
    // 如果之前优先级oldPri合法且不等于当前优先级curPri
    if(oldPri != LEVEL && oldPri != curPri) {
      // 调用remove()方法移除数据块
      remove(block, oldPri);
    }
    
    // 如果当前优先级curPri合法,通过当前优先级curPri从priorityQueues列表中获取对应数据块集合并将数据块添加进去
    if(curPri != LEVEL && priorityQueues.get(curPri).add(block)) {
      if(NameNode.blockStateChangeLog.isDebugEnabled()) {
        NameNode.blockStateChangeLog.debug(
          "BLOCK* NameSystem.UnderReplicationBlock.update:"
          + block
          + " has only "+ curReplicas
          + " replicas and needs " + curExpectedReplicas
          + " replicas so is added to neededReplications"
          + " at priority level " + curPri);
      }
    }
  }
        更新优先级update()方法用于当数据块副本数或期望副本数等发生变化时,调整数据块复制优先级,并调整其在UnderReplicatedBlocks中的相应存储位置。大体逻辑如下:

        1、首先搞清楚几个参数:curReplicas代表当前副本数,curReplicasDelta代表之前发生的副本数变化,curExpectedReplicas代表当前期望副本数,expectedReplicasDelta代表之前发生的期望副本数变化;

        2、计算之前的副本数oldReplicas和之前的期望副本数oldExpectedReplicas;

        3、计算当前的块复制优先级curPri;

        4、计算之前的块复制优先级oldPri;

        5、如果之前优先级oldPri合法且不等于当前优先级curPri:调用remove()方法移除数据块;

        6、如果当前优先级curPri合法,通过当前优先级curPri从priorityQueues列表中获取对应数据块集合并将数据块添加进去。

        未完待续,更多精彩尽在《HDFS源码分析之UnderReplicatedBlocks(二)》





相关文章
|
存储 缓存 分布式计算
|
存储 机器学习/深度学习 分布式计算
|
存储 调度 机器学习/深度学习
HDFS源码分析心跳汇报之数据块汇报
        在《HDFS源码分析心跳汇报之数据块增量汇报》一文中,我们详细介绍了数据块增量汇报的内容,了解到它是时间间隔更长的正常数据块汇报周期内一个smaller的数据块汇报,它负责将DataNode上数据块的变化情况及时汇报给NameNode。
1129 0
HDFS源码分析心跳汇报之DataNode注册
        HDFS源码分析心跳汇报之DataNode注册,近期推出!
899 0
|
机器学习/深度学习
HDFS源码分析心跳汇报之整体结构
        我们知道,HDFS全称是Hadoop Distribute FileSystem,即Hadoop分布式文件系统。既然它是一个分布式文件系统,那么肯定存在很多物理节点,而这其中,就会有主从节点之分。
1255 0
|
机器学习/深度学习
HDFS源码分析心跳汇报之数据结构初始化
        在《HDFS源码分析心跳汇报之整体结构》一文中,我们详细了解了HDFS中关于心跳的整体结构,知道了BlockPoolManager、BPOfferService和BPServiceActor三者之间的关系。
1144 0
|
存储 缓存
HDFS源码分析DataXceiver之读数据块
         在《HDFS源码分析DataXceiver之整体流程》一文中我们知道,无论来自客户端还是其他数据节点的请求达到DataNode时,DataNode上的后台线程DataXceiverServer均为每个请求创建一个单独的后台工作线程来处理,这个工作线程就是DataXceiver。
1232 0
|
存储 分布式计算 Hadoop
HDFS源码分析DataXceiver之整体流程
        在《HDFS源码分析之DataXceiverServer》一文中,我们了解到在DataNode中,有一个后台工作的线程DataXceiverServer。它被用于接收来自客户端或其他数据节点的数据读写请求,为每个数据读写请求创建一个单独的线程去处理。
1582 0

热门文章

最新文章