Hdfs NameNode中数据块、数据节点及副本状态概述_hdfs中的节点数和副本数的联系

       HDFS集群以Master-Slave模式运行，主要有两类节点：一个Namenode(即Master)和多个Datanode(即Slave)。

​

       在hdfs文件系统中，NameNode是HDFS中的主节点，其主要管理和维护hdfs文件系统中的两个重要关系；

数据块和节点管理

     名字节点维护着HDFS文件系统中两个重要的关系：

1. HDFS文件系统的命名空间（namespace）；包括文件系统目录树，文件/目录信息以及文件对应的数据块索引，即每个文件对应的数据块列表；这个信息保存在INodeFile.blocks字段中。
2. 数据块与数据节点的对应关系，即某一数据块副本保存在哪些数据节点的信息；这个信息是由DataNode启动时，由DataNode上报给NameNode的，也就是说这部分信息是NameNode动态构建起来的，而不是从fsimage加载的。

     在第二关系中，fsimage镜像文件中不会记录数据块保存在哪些数据节点的信息。而是在数据节点DataNode加入到集群时，由数据节点blockReport向NameNode提供他所包含的数据块副本动态建立起来的。和第一关系相同的是，随着系统的运行，数据块和数据节点都会发生变化，上述对应关系也就会不断地发生变化，名字节点必须跟踪、维护这个关系，以保证系统的正常工作。

相关类

数据块关系中涉及的基本类如下：

1. BlocksMap--数据块映射图类，名字节点上数据块的元数据信息，其保存了数据块到数据节点列表以及所属INode文件节点的映射关系
2. BlocksMap.BlockInfo--BlocksMap的内部类，保存了block的数据信息类，其主要保存了
   1. 数据块-->文件(INodeFile)
   2. 数据块-->数据节点(DatanodeDescriptor)
3. DatanodeDescriptor--数据节点类，是对数据节点的管理抽象，里面包含了许多与数据节点有关的变量信息
4. DatanodeID和DatanodeInfo--数据节点基本类，是DatanodeDescriptor的父类，在这里定义了一个节点最基本的信息
5. GSet以及LightWeightGSet--链表集合类，保存了blockInfo信息的集合类，采用了哈希存储的方式存储block数据块信息
6. PendingReplicationBlocks和UnderReplicatedBlocks--副本相关类
7. FSNamesystem--命名系统空间门面类，其内部维护了NameNode中多个组件间的交互处理，共同对外提供了对hdfs文件的管理和操作

BlocksMap

        blocksMap管理着名字节点上数据块的元数据，包括数据块所属的INode和那些数据节点保存数据块等信息。也就是说，如果要定位某个数据块在哪些数据节点存储，只要访问名字节点的BlocksMap对象。BlocksMap采用org.apache.hadoop.hdfs.uti.GSet，一个比较特殊的集合类型，存放(对象blocks)名字节点管理的所有数据块。GSet特殊的地方在于它是个集合，但提供了类似映射的功能，如blocks.get()，可以根据数据块对象获得它对应的BlockInfo对象。

        BlockInfo(数据块信息)是BlocksMap的内部类，也是Block的子类，它增加了数据块所属的INode信息和保存该数据块的数据节点信息。INode信息保存在BlockInfo.inode中，但数据块所在数据节点DatanodeDescriptor对象的保存就比较特殊了，它保存在数据类型为Object的数组triplets中，而不是保存在一个DatanodeDescriptor数组中。而且数组triplets除了保存数据节点信息(第i个数据节点的信息保存在元素tripltes[3*i]中)，还以双向链表的形式保存该数据节点上其他两个数据块的数据块信息对象，即triplets[3*i+1]保存了当前数据块的前一个数据块的BlockInfo，triplets[3*i+2]保存了后面一个数据块的BlockInfo。也就是说，沿着triplets[3*i+1]或triplets[3*i+2]，可以遍历某个数据节点拥有的所有数据块的BlockInfo信息。代码如下：

class BlocksMap {
  /** Constant {@link LightWeightGSet} capacity. */
  private final int capacity;
  
  private GSet<Block, BlockInfo> blocks; // 保存block对应的blockInfo元数据
 
  BlocksMap(int capacity) {
    // Use 2% of total memory to size the GSet capacity
    this.capacity = capacity;
    // 采用轻量级的GSet实现block与blockInfo的映射存储
    this.blocks = new LightWeightGSet<Block, BlockInfo>(capacity) {
      @Override
      public Iterator<BlockInfo> iterator() {
        SetIterator iterator = new SetIterator();
        iterator.setTrackModification(false);
        return iterator;
      }
    };
  }
  // ......
}

static class BlockInfo extends Block implements LightWeightGSet.LinkedElement {
    // 数据块信息所属的INode文件节点 BlockCollection接口的实现就是INodeFile
    private BlockCollection bc;
 
    /** For implementing {@link LightWeightGSet.LinkedElement} interface */
    private LightWeightGSet.LinkedElement nextLinkedElement;
 
    /**
     * This array contains triplets of references.
     * For each i-th data-node the block belongs to
     * triplets[3*i] is the reference to the DatanodeDescriptor
     * and triplets[3*i+1] and triplets[3*i+2] are references 
     * to the previous and the next blocks, respectively, in the 
     * list of blocks belonging to this data-node.
     * triplets对象数组保存同一数据节点上的连续的block块。triplets[3*i]保存的是当前的数据节点
     * triplets[3*i+1]和triplets[3*i+2]保存的则是一前一后的block信息
     */
    private Object[] triplets; // 数据块所属的DataNode信息
    // ......
}

       BlockInfo的triplets[3*i]保存数据节点信息，类型为DatanodeDescriptor，它是DatanodeInfo的子类，而DatanodeInfo又是DatanodeID的子类，同时DatanodeDescriptor也拥有一些内部类。通过DatanodeInfo可以定位一个数据节点，并了解该节点的一些状态，在数据节点和名字节点、客户端和名字节点的IPC接口中都有DatanodeInfo的应用。

DatanodeDescriptor

       和BlockInfo一样，DatanodeDescriptor是对数据节点管理的抽象，其内部定义了1、数据块到数据节点的映射关系，2、并且添加对数据节点状态的管理以及3、用于在下一次心跳信息中产生名字节点指令的数据块列表集合：

1、数据节点的状态

        包括isAlive、firstBlockReport和decommissiongingStatus、currApproxBlocksScheduled等。其中decommissioningStatus(类型为DatanodeDescriptor.DecommissionStatus)比较特殊，它只在节点处于撤销状态时使用；currApproxBlocksScheduled等几个成员变量用于估计数据节点的负载，为写文件操作分配数据块或进行数据块复制时，会根据节点负债信息，优先选择比较空闲的节点作为目标。

2、用于产生到该数据节点的名字节点的指令

        包括bandwidth、replicateBlocks、recoverBlocks和invalidateBlocks四个成员变量，它们分别用于产生均衡器工作带宽更新(DNA_BALANCERBANDWIDTHUPDATE)、数据块复制(DNA_TRANSFER)、数据块恢复(DNA_RECOVERBLOCK)和数据块删除(DNA_INVALIDATE)的名字节点指令，以数据删除为例，成员变量invalidateBlocks保存了这个数据节点上等待删除的数据块，在下次心跳中，名字节点可能会产生一个DNA_INVALIDAATE操作，让数据节点删除集合invalidateBlocks中的一些数据块。成员变量replicateBlocks和recoverBlocks的类型是数据块队列BlockQueue，它的元素类型是BlockTargetPair。如类名所示，BlockTargeParis包含了数据块和目标数据节点(列表)两项信息，对应复制，目标数据节点是复制的目标节点，对于数据块恢复，目标数据节点则是参与恢复过程的数据节点。

3、成员变量blockList

        它是该数据节点保存的数据块队列的头节点，队列元素类型为BlocksMap.BlockInfo

// DatanodeDescriptor数据节点描述类跟踪描述了一个数据节点的状态信息
public class DatanodeDescriptor extends DatanodeInfo {
  
  // Stores status of decommissioning.
  // If node is not decommissioning, do not use this object for anything.
  DecommissioningStatus decommissioningStatus = new DecommissioningStatus();
 
  /** Block and targets pair */
  // 数据块以及目标数据节点列表映射类
  public static class BlockTargetPair {
    public final Block block; // 目标数据块
    public final DatanodeDescriptor[] targets; // 该数据块的目标数据节点  
 
    BlockTargetPair(Block block, DatanodeDescriptor[] targets) {
      this.block = block;
      this.targets = targets;
    }
  }
 
  /** A BlockTargetPair queue. */
  private static class BlockQueue { // block块目标数据节点类队列; 此类维护了BlockTargetPair列表对象
    private final Queue<BlockTargetPair> blockq = new LinkedList<BlockTargetPair>();
    // ......
  }
  
  // 支持异构特性的storage存储描述；从之前的 数据块与数据节点关系 变成 数据块与storage存储
  // (保存了数据块与storage存储的对应关系)
  private final Map<String, DatanodeStorageInfo> storageMap = 
      new HashMap<String, DatanodeStorageInfo>();
  
  // isAlive == heartbeats.contains(this)
  // This is an optimization, because contains takes O(n) time on Arraylist
  protected boolean isAlive = false;
  protected boolean needKeyUpdate = false;
 
  // A system administrator can tune the balancer bandwidth parameter
  // (dfs.balance.bandwidthPerSec) dynamically by calling
  // "dfsadmin -setBalanacerBandwidth <newbandwidth>", at which point the
  // following 'bandwidth' variable gets updated with the new value for each
  // node. Once the heartbeat command is issued to update the value on the
  // specified datanode, this value will be set back to 0.
  private long bandwidth;
 
  /** A queue of blocks to be replicated by this datanode */
  // 此数据节点上待复制的block块列表
  private BlockQueue replicateBlocks = new BlockQueue();
  /** A queue of blocks to be recovered by this datanode */
  // 此数据节点上待租约恢复的块列表
  private BlockQueue recoverBlocks = new BlockQueue();
  /** A set of blocks to be invalidated by this datanode */
  // 此数据节点上无效待删除的块列表
  private Set<Block> invalidateBlocks = new TreeSet<Block>();
 
  // ......
}

public class DatanodeStorageInfo { // 每个目录对应一个实例，其具体存储可以是异构的
  private final DatanodeDescriptor dn;
  private final String storageID;
  private StorageType storageType;
  private State state;
 
  // 每个存储目录的基本使用信息
  private long capacity;
  private long dfsUsed;
  private volatile long remaining;
  private long blockPoolUsed;
 
  // 记录当前存储上保存的数据块副本信息
  private volatile BlockInfo blockList = null; // 
  // ......
}

针对DatanodeStorageInfo的一些操作方法如下：

       当数据节点成功接收到一个新的数据块副本后，会通过远程方法blockReceived()报告名字节点。名字节点自然需要把它添加/更新到对应数据节点DatanodeDescriptor中的DatanodeStorageInfo对象(实际存储对象，支持异构)中，使用的就是blockManager.addBlock()方法。它首先调用BlockInfo.addStorage()将对象自己，也就是DatanodeStorageInfo添加到数据块所属数据节点DataNode列表中，也就是BlockInfo对象中的triplets[]数组中。接着，又调用BlockInfo的listInsert()方法，将数据块插入到数据节点DataNode管理的数据块列表中。也就是将BlockInfo对象加入到DataNode存储对应的DatanodeStorageInfo对象中的blockList链表中。代码如下：

// DatanodeStorageInfo#addBlock()
public boolean addBlock(BlockInfo b) {
  // First check whether the block belongs to a different storage
  // on the same DN.  首先检查数据块是否是同一个DN上的另一个存储
  boolean replaced = false;
  DatanodeStorageInfo otherStorage =
      b.findStorageInfo(getDatanodeDescriptor());
 
  if (otherStorage != null) {
    if (otherStorage != this) {
      // The block belongs to a different storage. Remove it first.
      otherStorage.removeBlock(b);
      replaced = true;
    } else {
      // The block is already associated with this storage.
      return false;
    }
  }
 
  // add to the head of the data-node list
  // 首先将当前存储添加到数据块所属的存储列表中
  b.addStorage(this);
  // 将当前数据块添加到存储管理的数据块列表blockList中
  blockList = b.listInsert(blockList, this);
  numBlocks++;
  return !replaced;
}

        DatanodeDescriptor.getReplicationCommand()用于产生一条数据块复制指令，它在队列DatanodeDescriptor.replicateBlocks中获取最多maxTransfers个元素，然后根据这些元素构造BlockCommand对象。这个方法很简单，在DatanodeDescriptor中，类似的方法还有getLeaseRecoverCommand()、getInvalidateBlocks()等；生成对应的数据块复制指令的函数调用栈如下(主要关注DatanodeDescriptor中的方法)：

1. namesystem.handleHeartbeat() // FSNamesystem是名字节点的系统门面类，由它对内部组件交互进行统一操作
   1. // get datanode commands
      DatanodeCommand[] cmds = blockManager.getDatanodeManager().handleHeartbeat()；
      1. DatanodeManager.handleHeartbeat()；
         1. DatanodeDescriptor nodeinfo = getDatanode(nodeReg)；
            // check pending replication
            List<BlockTargetPair> pendingList = nodeinfo.getReplicationCommand( maxTransfers)；
            if (pendingList != null) {
                  cmds.add(new BlockCommand(DatanodeProtocol.DNA_TRANSFER, blockPoolId, pendingList));
            }

public class DatanodeDescriptor extends DatanodeInfo {
  // 在DatanodeDescriptor中只是简单的返回需要复制的数据块列表
  public List<BlockTargetPair> getReplicationCommand(int maxTransfers) {
    return replicateBlocks.poll(maxTransfers);
  }
  // ......
}

数据块副本状态

       BlocksMap和DatanodeDescriptor一起保存名字节点第二关系，即数据块和数据节点的对应关系，在hdfs文件系统中，BlocksMap是由BlockManager进行统一管理的；包括名字节点上对数据块副本的操作管理，下面是BlockManager中涉及数据块管理的成员变量：

public class BlockManager {
  private volatile long pendingReplicationBlocksCount = 0L;
  private volatile long corruptReplicaBlocksCount = 0L;
  private volatile long underReplicatedBlocksCount = 0L;
  private volatile long scheduledReplicationBlocksCount = 0L;
  
  private final Namesystem namesystem;
  
  /**
   * Mapping: Block -> { BlockCollection, datanodes, self ref }
   * Updated only in response to client-sent information.
   */
  final BlocksMap blocksMap;
  
  /** Replication thread. */
  final Daemon replicationThread = new Daemon(new ReplicationMonitor());
  
  /** Store blocks -> datanodedescriptor(s) map of corrupt replicas */
  // 存储损坏的数据块副本集合
  final CorruptReplicasMap corruptReplicas = new CorruptReplicasMap();
  
  /** Blocks to be invalidated. */
  // 待删除的数据块副本集合
  private final InvalidateBlocks invalidateBlocks;
  
  // 推迟操作的数据块副本集合
  private final Set<Block> postponedMisreplicatedBlocks = Sets.newHashSet();
 
  // 多余的数据块副本集合
  public final Map<String, LightWeightLinkedSet<Block>> excessReplicateMap =
    new TreeMap<String, LightWeightLinkedSet<Block>>();
    
  // 等待复制的数据块副本集合
  public final UnderReplicatedBlocks neededReplications = new UnderReplicatedBlocks();
 
  // 已经生成复制请求的数据块副本
  final PendingReplicationBlocks pendingReplications;
}

       基本上对特定数据块副本的下一步操作，BlockManager都有一个成员变量保存着相应的数据块和执行操作时需要的附加信息，有些还有计时器存放等待操作的数据块数目，具体内容如下：

• corruptReplicas：保存损坏的数据块副本，如数据节点的数据块扫描器发现的错误数据块副本。
• invalidateBlocks：保存待删除的数据块副本集合。等待删除的数据块副本来自corruptReplicas和excessReplicateMap这两个集合，加入这个队列中的数据块副本会由NameNode通过名字节点指令向DataNode下发删除指令。
• excessReplicateMap：多余副本，减少文件的副本数(可通过“hadoop fs -setrep”命令)会产生多余副本，名字节点启动时也可能发现多余副本，多余副本由名字节点在数据块的多个副本中选择得到。
• neededReplications：等待复制，准备生成复制请求的数据块副本，数据块复制是为了让数据块的副本数满足文件的副本系数。
• pendingReplications：已经生产复制请求的数据块副本保存在pendingReplications中，也就是说，当复制请求生成后，数据块的信息会从neededReplications取出，放入pendingReplications中。

接下来只对部分副本状态集合进行分析，其余的可以参照blockManager中的源代码进行分析：       

        上面列表中的excessReplicateMap的类型比较简单，是一个数据节点存储标识到一组数据块对象(Collection<Block>)的映射。excessReplicateMap使用java.util.TreeMap作为具体的映射实现。其他变量如corruptReplicas、neededReplications和pendingReplications都定义了对应的类来保存需要的数据。

        BlockManager.corruptReplicas的类型是CorruptReplicasMap，保存数据节点发现的损坏数据块副本。当客户端发现有损坏的数据块时会通过ClientProtocol.reportBadBlocks()方法向NameNode汇报损坏的数据块副本信息，之后数据节点会通过DataNodeProtocol.reportBadBlocks()方法汇报损坏的数据块副本，之后BlockManager会将损坏的数据块副本添加到当前的corruptReplicas数据结构中。由于数据块往往有多个副本，只有该数据块的所有副本都损坏了，系统才认为数据块损坏，当部分副本出现损坏时，名字节点会进行数据块复制，复制完好的数据块副本直至数据块副本数恢复正常。CorruptReplicasMap用于配合上述过程，它通过成员变量corruptReplicasMap，记录了损坏副本以及副本所在的(一组)数据节点，当发现新的损坏副本是，通过addToCorruptReplicasMap()方法添加副本信息到CorruptRelicasMap对象中；当损坏数据块从数据节点上删除时，则调用removeFromCorruptReplicasMap()删除记录。CorruptReplicasMap的实现源代码如下：

public class CorruptReplicasMap{
 
  /** The corruption reason code */
  // 数据块副本损坏的原因
  public static enum Reason {
    NONE,                // not specified.
    ANY,                 // wildcard reason
    GENSTAMP_MISMATCH,   // mismatch in generation stamps
    SIZE_MISMATCH,       // mismatch in sizes
    INVALID_STATE,       // invalid state
    CORRUPTION_REPORTED  // client or datanode reported the corruption
  }
 
  // 底层的存储结构
  private final SortedMap<Block, Map<DatanodeDescriptor, Reason>> corruptReplicasMap =
    new TreeMap<Block, Map<DatanodeDescriptor, Reason>>();
 
  void addToCorruptReplicasMap(Block blk, DatanodeDescriptor dn,
      String reason, Reason reasonCode) {
    Map <DatanodeDescriptor, Reason> nodes = corruptReplicasMap.get(blk);
    if (nodes == null) {
      nodes = new HashMap<DatanodeDescriptor, Reason>();
      corruptReplicasMap.put(blk, nodes);
    }
    
    // 添加对应的reason
    // Add the node or update the reason.
    nodes.put(dn, reasonCode);
  }
 
  void removeFromCorruptReplicasMap(Block blk) {
    if (corruptReplicasMap != null) {
      corruptReplicasMap.remove(blk);
    }
  }
}

其最终所有的数据块副本状态转换图如下：