第七章：小朱笔记hadoop之源码分析-hdfs分析 第三节：hdfs实现分析

第七章：小朱笔记hadoop之源码分析-hdfs分析

第三节：hdfs实现分析

3.3 namenode

（1）FSDirectory

FSDirectory用来管理HDFS整个文件系统的namespace也就是整个目录树.

HDFS是会将namespace保存到namenode的本地文件系统的fsimage的文件中。从该文件中，namenode每次重启都能将整个HDFS的namespace重新初始化。所以FSDirectory负责对 fsimage相关操作。

HDFS的各种操作，namenode都会记录相应的操作日志(editlog文件)，以便周期性的将该日志于fsimage进行合并，生成新的 fsimage 。所以FSDirectory负责对editlog的相关操作。

HDFS整个文件系统的namespace在namenode的内存中，是以树的结构来维护的。在HDFS中，不管是目录还是文件，都被看作是 INode，如果是目录，则其实际类标识为INodeDirectory，如果是文件，则其相应的类标识为INodeFile，这两个类都是INode的派生类。INodeDirectory中包含一个成员数组变量List<INode> children，如果该目录下有子目录或者文件，其子目录或文件的引用就会保存在children列表中。HDFS就是通过这种方式来维持整个文件系统的目录结构。

（2）FsImage

       Namenode会将HDFS的文件/目录的元数据存储在一个叫fsimage的二进制文件中，每次保存fsimage之后到下次保存之间的所有 hdfs操作，将会记录在editlog文件中，当editlog达到一定的大小（bytes，由fs.checkpoint.size参数定义）或从上次保存过后一定时间段过后（sec，由fs.checkpoint.period参数定义），namenode会重新将内存中对整个HDFS的目录树和文件元数据刷到fsimage文件中。Namenode就是通过这种方式来保证HDFS中元数据信息的安全性。Fsimage是一个二进制文件，当中记录了HDFS中所有文件和目录的元数据信息。

当namenode重启加载fsimage时，就是按照如下格式协议从文件流中加载元数据信息。从fsimag的存储格式可以看出，fsimage保存有如下信息：

    1）image head
       imgVersion(int)：当前image的版本信息
       namespaceID(int)：用来确保别的HDFS instance中的datanode不会误连上当前NN。
       numFiles(long)：整个文件系统中包含有多少文件和目录
       genStamp(long)：生成该image时的时间戳信息。
    2）每个文件或目录的源数据信息，如果是目录，则包含以下信息：
       path(String)：该目录的路径，如”/user/build/build-index”
       replications(short)：副本数（目录虽然没有副本，但这里记录的目录副本数也为3）
       mtime(long)：该目录的修改时间的时间戳信息
       atime(long)：该目录的访问时间的时间戳信息
       blocksize(long)：目录的blocksize都为0
       numBlocks(int)：实际有多少个文件块，目录的该值都为-1，表示该item为目录
       nsQuota(long)：namespace Quota值，若没加Quota限制则为-1
       dsQuota(long)：disk Quota值，若没加限制则也为-1
       username(String)：该目录的所属用户名
       group(String)：该目录的所属组
       permission(short)：该目录的permission信息，如644等，有一个short来记录。
    3）若从fsimage中读到的item是一个文件，则还会额外包含如下信息：
       blockid(long)：属于该文件的block的blockid，
       numBytes(long)：该block的大小
       genStamp(long)：该block的时间戳

    当该文件对应的numBlocks数不为1，而是大于1时，表示该文件对应有多个block信息，此时紧接在该fsimage之后的就会有多个blockid，numBytes和genStamp信息。因此，在namenode启动时，就需要对fsimage按照如下格式进行顺序的加载，以将fsimage中记录的HDFS元数据信息加载到内存中。

（3）BlocksMap

      从以上fsimage中加载如namenode内存中的信息中可以很明显的看出，在fsimage中，并没有记录每一个block对应到哪几个 datanodes的对应表信息，而只是存储了所有的关于namespace的相关信息。而真正每个block对应到datanodes列表的信息在 hadoop中并没有进行持久化存储，而是在所有datanode启动时，每个datanode对本地磁盘进行扫描，将本datanode上保存的 block信息汇报给namenode，namenode在接收到每个datanode的块信息汇报后，将接收到的块信息，以及其所在的datanode 信息等保存在内存中。HDFS就是通过这种块信息汇报的方式来完成 block -> datanodes list的对应表构建。Datanode向namenode汇报块信息的过程叫做blockReport，而namenode将block -> datanodes list的对应表信息保存在一个叫BlocksMap的数据结构中。

      BlocksMap实际上就是一个Block对象对BlockInfo对象的一个Map表，其中Block对象中只记录了 blockid，block大小以及时间戳信息，这些信息在fsimage中都有记录。而BlockInfo是从Block对象继承而来，因此除了 Block对象中保存的信息外，还包括代表该block所属的HDFS文件的INodeFile对象引用以及该block所属datanodes列表的信息。
      因此在namenode启动并加载fsimage完成之后，实际上BlocksMap中的key，也就是Block对象都已经加载到 BlocksMap中，每个key对应的value(BlockInfo)中，除了表示其所属的datanodes列表的数组为空外，其他信息也都已经成 功加载。所以可以说：fsimage加载完毕后，BlocksMap中仅缺少每个块对应到其所属的datanodes list的对应关系信息。所缺这些信息，就是通过上文提到的从各datanode接收blockReport来构建。当所有的datanode汇报给namenode的blockReport处理完毕后，BlocksMap整个结构也就构建完成。

class BlocksMap {
        
  /**
   * Internal class for block metadata.
   */
  static class BlockInfo extends Block implements LightWeightGSet.LinkedElement {

    /** 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.
     * 
     *  该数组包含了块的三组引用。具体信息描述如下：
     *  对于块所属的第i个Datanode，数组元素triplets[3*i]引用了描述该Datanode结点的DatanodeDescriptor实例，
     *  数组元素triplets[3*i+1]与triplets[3*i+2]引用的分别是前（previous）一个块与后（next）一个块。
     *  注意，triplets[3*i+1]与triplets[3*i+2]所引用的块应该在指定的同一个Datanode上的块列表中
     */
    private INodeFile          inode; // 文件结点  
    
    private Object[] triplets; // 文件的块列表  

    BlockInfo(Block blk, int replication) {
      super(blk);
      this.triplets = new Object[3*replication];
      this.inode = null;
    }

    INodeFile getINode() {
      return inode;
    }

  .............
  }
  
  private GSet<Block, BlockInfo> blocks;//Block找对应的文件和这个Block存放的DataNode的相关信息

  BlocksMap(int initialCapacity, float loadFactor) {
    this.capacity = computeCapacity();
    this.blocks = new LightWeightGSet<Block, BlockInfo>(capacity);
  }

（4）BlockInfo中datanode列表数据结构

     在BlockInfo中，将该block所属的datanodes列表保存在一个Object[]数组中.
     Namenode采用这种结构来保存block->datanode list的目的在于节约namenode内存。由于namenode将block->datanodes的对应关系保存在了内存当中，随着HDFS 中文件数的增加，block数也会相应的增加，namenode为了保存block->datanodes的信息已经耗费了相当多的内存，如果还像 这种方式一样的保存datanode->block list的对应表，势必耗费更多的内存，而且在实际应用中，要查一个datanode上保存的block list的应用实际上非常的少，大部分情况下是要根据block来查datanode列表，所以namenode中通过上图的方式来保存 block->datanode list的对应关系，当需要查询datanode->block list的对应关系时，只需要沿着该数据结构中next Block的指向关系，就能得出结果，而又无需保存datanode->block list在内存中。

 （5）CorruptReplicationMap
      保存在datanodes上的block，常常会因为各种各样的原因（磁盘损坏，校验错误等）出错，这种情况下，namenode中会将这些datanode上的block标记为Corrupt状态，并记录该 block在哪几台datanode上保存的副本是corrupt的。这些信息，就是保存在CorruptReplicationMap中。 CorruptReplicationMap内部其实就是一个Block-> Collection<DatanodeDescriptor>的TreeMap，namenode中通过该数据结构来记录哪些block的哪(几)个副本为corrupt状态，并会在随后的将这些记录的block放入到一个 recentInvalidateSets（Map<String, Collection<Block>>）中。
      在datanode的每次heartbeat过来namenode的时候，namenode就有机会查询该 recentInvalidateSets，看是否在该汇报的datanode上有被标记为invalidate的block，如果有，就在 heartbeat中返回给该datanode一个cmd，告诉该datanode将该有问题的block进行删除。

  //
  // Store blocks-->datanodedescriptor(s) map of corrupt replicas
  //
  //保存损坏（如：校验没通过）的数据块到对应DataNode的关系，CorruptReplicasMap类图如下，类只有一个成员变量，
  //保存Block到一个DatanodeDescriptor的集合的映射和这个映射上的一系列操作
  //
  public CorruptReplicasMap corruptReplicas = new CorruptReplicasMap();


**
 * Stores information about all corrupt blocks in the File System.
 * A Block is considered corrupt only if all of its replicas are
 * corrupt. While reporting replicas of a Block, we hide any corrupt
 * copies. These copies are removed once Block is found to have 
 * expected number of good replicas.
 * Mapping: Block -> TreeSet<DatanodeDescriptor> 
 * 
 * 该类是有关失效块的映射表。如果一个块被认为是失效的，是指该块对应的全部块副本都失效而不可用
 */

public class CorruptReplicasMap{

  private Map<Block, Collection<DatanodeDescriptor>> corruptReplicasMap =
    new TreeMap<Block, Collection<DatanodeDescriptor>>();
    ........
}

 

（6）UnderReplicatedBlocks
        HDFS中的block，通常会对应多个副本，假设某block的副本数是3，那么并不一定在任何时候，该block的可用副本数都能保持为3。原因是集群中随时都有可能会有硬盘损坏，datanode下线等原因。发生这种情况时，某些block的对应的副本就会下降。HDFS将副本数没有达到其期望值的block引用保存在UnderReplicatedBlocks数据结构中。UnderReplicatedBlocks其实是一个list列 表，其结构为：ArrayList<TreeSet<Block>>。List中的每一个item其实都是一个 TreeSet，set中是一系列的block引用。UnderReplicatedBlocks将该list中的block按照优先级来分类。优先级由 该block当前副本数和缺少的副本数来决定，也就是说，缺少的副本数越多，标识该block的副本被拷贝到其他datanode上的请求就越紧急，其优 先级也就越高。因此，优先级为0的blocks，就保存在优先级为0的TreeSet中，以此类推。

    /**
     * Store set of Blocks that need to be replicated 1 or more times. Set of:
     * Block
     * 
     * 需要进行复制的数据块。UnderReplicatedBlocks的类图如下，它其实是一个数组，数组的下标是优先级（0的优先级最高，
     * 如果数据块只有一个副本，它的优先级是0），
     * 数组的内容是一个Block集合。UnderReplicatedBlocks提供一些方法，对Block进行增加，修改，查找和删除。
     * 
     */
    private UnderReplicatedBlocks neededReplications = new UnderReplicatedBlocks();


    class UnderReplicatedBlocks implements Iterable<Block> {
     static final int LEVEL = 3;
     //优先级队列，为一个二维集合，在NN检查到块副本不足的情况下，根据副本数量的等情况，将块加入到相应优先级列表中；
     private List<TreeSet<Block>> priorityQueues = new ArrayList<TreeSet<Block>>();
     ........
  }
      

   

（7）PendingReplicationBlocks 
      当namenode发现某些block的可用副本数低于其期望副本数时，在一定时期内就会对该block进行副本拷贝的操作，以将其可用副本数提高 到期望副本数（通常为3）。而正在进行副本拷贝的block，namenode会将其引用保存在PendingReplicationBlocks中，当该block从一台src datanode拷贝到target datanode的副本拷贝操作时，就会将该block的引用保存在PendingReplicationBlocks中。
      在block做副本拷贝时，有些能够很正常的拷贝成功，但是有些拷贝操作因为各种各样的原因，会发生失败，或者超时，默认的超时时间是5分钟。当发生超时时，namenode会将这些block重新添加到neededReplicationBlocks中，以便接下来重新对其进行副本拷贝操作。 PendingReplicationBlocks中还有一个后台线程，就是这个线程在不停的检查正在做副本拷贝的block是否有超时的现象

/***************************************************
 * PendingReplicationBlocks does the bookkeeping of all
 * blocks that are getting replicated.
 *
 * It does the following:
 * 1)  record blocks that are getting replicated at this instant.
 * 2)  a coarse grain timer to track age of replication request
 * 3)  a thread that periodically identifies replication-requests
 *     that never made it.
 *     
 *     该类org.apache.hadoop.hdfs.server.namenode.PendingReplicationBlocks是描述当前尚未完成块副本复制的块的列表。
 *     因为HDFS支持块的流水线复制，当文件系统客户端开始向第一个Datanode复制数据块的时候，会在第一个Datanode上启动复制进程，将该结点上接收到的（部分）
 *     数据块开始向第二个Datanode上传输复制，而第二个Datanode结点又向第三个Datanode结点进行流水线复制，……，直到满足副本因子要求。
 *     所以，在执行流水线复制的过程中，因为这种可并行的复制方式使得某些块副本的复制完成时间呈现阶梯状，也就是使用一个数据结构来保存这些尚未复制完成的块副本
 *     
 *     pendingReplications保存了所有正在进行复制的数据块，使用Map是需要一些附加的信息PendingBlockInfo。这些信息包括时间戳，
 *     用于检测是否已经超时，和现在进行复制的数目numReplicasInProgress。timedOutItems是超时的复制项，
 *     超时的复制项在FSNamesystem的processPendingReplications方法中被删除，并重新复制。timerThread是用于检测复制超时的线程的句柄，
 *     对应的线程是PendingReplicationMonitor的一个实例，它的run方法每隔一段会检查是否有超时的复制项，如果有，将该数据块加到timedOutItems中。Timeout是run方法的检查间隔，defaultRecheckInterval是缺省值
 *     
 *
 ***************************************************/
class PendingReplicationBlocks {
  private Map<Block, PendingBlockInfo> pendingReplications;

/*  数据块集合，在PendingReplicationMonitor线程内，会定时扫描pendingReplications维护的集合，检查到某块冗余时间已经超时，
  会将该数据块块加入timedOutItems集合；在ReplicationMonitor中，又会将timedOutItems中的数据块重新加入neededReplications；*/
  private ArrayList<Block> timedOutItems;
  Daemon timerThread = null;
  private volatile boolean fsRunning = true;

  //
  // It might take anywhere between 5 to 10 minutes before
  // a request is timed out.
  //
  private long timeout = 5 * 60 * 1000;
  private long defaultRecheckInterval = 5 * 60 * 1000;

  ........
  }

（8）datanodeMap

 它保存了StorageID -> DatanodeDescriptor的映射，用于保证DataNode使用的Storage的一致性。

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

（9）DatanodeDescriptor

DatanodeDescriptor更进一步，包含了DataNode上一些Block的动态信息。DatanodeDescriptor包含了内部类BlockTargetPair，它保存Block和对应DatanodeDescriptor的关联，BlockQueue是BlockTargetPair队列。 DatanodeDescriptor包含了两个BlockQueue，分别记录了该DataNode上正在复制（replicateBlocks）和Lease恢复（recoverBlocks）的Block,同时还有一个Block集合，保存的是该DataNode上已经失效的Block。DatanodeDescriptor提供一系列方法，用于操作上面保存的队列和集合。也提供get*Command方法，用于生成发送到DataNode的命令。当NameNode收到DataNode对现在管理的Block状态的汇报是，会调用reportDiff，找出和现在NameNode上的信息差别，以供后续处理用。

/**************************************************
 * DatanodeDescriptor tracks stats on a given DataNode,
 * such as available storage capacity, last update time, etc.,
 * and maintains a set of blocks stored on the datanode. 
 *
 * This data structure is a data structure that is internal
 * to the namenode. It is *not* sent over-the-wire to the Client
 * or the Datnodes. Neither is it stored persistently in the
 * fsImage.
 * 
 **************************************************/
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();

  ............
   
  private volatile BlockInfo blockList = null;
  // 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 */
  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>();

  /* Variables for maintaning number of blocks scheduled to be written to
   * this datanode. This count is approximate and might be slightly higger
   * in case of errors (e.g. datanode does not report if an error occurs 
   * while writing the block).
   */
  private int currApproxBlocksScheduled = 0;
  private int prevApproxBlocksScheduled = 0;
  private long lastBlocksScheduledRollTime = 0;
  private static final int BLOCKS_SCHEDULED_ROLL_INTERVAL = 600*1000; //10min
  
  // Set to false after processing first block report
  private boolean firstBlockReport = true; 
  
  .........
 }

（10）excessReplicateMap

保存Datanode上有效但需要删除的数据块（StorageID ->TreeSet<Block>）比如一个Datanode故障恢复后，上面的数据块在系统中副本数太多，需要删除一些数据块。

    Map<String, Collection<Block>> excessReplicateMap = new TreeMap<String, Collection<Block>>();

（11）heartbeats

通过心跳信息保存所有目前活着的DataNode，线程HeartbeatMonitor会定期检查

    /**
     * Stores a set of DatanodeDescriptor objects. This is a subset of
     * {@link #datanodeMap}, containing nodes that are considered alive. The
     * {@link HeartbeatMonitor} periodically checks for outdated entries, and
     * removes them from the list. 所有目前活着的DataNode，线程HeartbeatMonitor会定期检查
     */
    ArrayList<DatanodeDescriptor> heartbeats = new ArrayList<DatanodeDescriptor>();

类图: