Hadoop的两大系统之一“HDFS”。所谓分布,是说整个文件系统的内容并非集中存储在一台或几台“文件服务器上”,而是分散在集群的不同节点上。理想的情景是集群内的每一台机器都承担着一些内容的存储。HDFS 是 Hadoop 集群的文件系统,这是一种分布( distributed )、容错( faulttolerant )的文件系统。对于大数据处理系统,文件系统之所以应该是分布式的,不再仅仅是因为容量和容错的问题。大数据处理需要并行化,需要把计算量分摊到很多节点上,但是,如果海量的数据都集中在一起,需要在计算的时候通过网络传输过去,那就不大现实了,因为这个集中点及其周边的网络显然会成为瓶颈。所以大数据处理有个原则,就是数据在哪里,计算就在哪里。

1. NameNode

HDFS 的结构是一种主/从结构,负责提供查名服务的 NameNode 扮演着主节点的角色。

public class NameNode implements NameNodeStatusMXBean {

protected FSNamesystem namesystem; //目录系统
protected NameNodeHttpServer httpServer; //为通过 HTTP 浏览器管理 HDFS 提供 Web 服务
private NameNodeRpcServer rpcServer; //提供 RPC 服务

//启动入口
public static void main(String argv[]) throws Exception {
    try {
      StringUtils.startupShutdownMessage(NameNode.class, argv, LOG);
      NameNode namenode = createNameNode(argv, null);
      if (namenode != null) {
        namenode.join();
      }
    }
  }

// 创建NameNode, 根据命令行选项执行格式化，版本回滚等操作
public static NameNode createNameNode(String argv[], Configuration conf)
      throws IOException {
    //解析执行参数
    // Parse out some generic args into Configuration.
    GenericOptionsParser hParser = new GenericOptionsParser(conf, argv);
    argv = hParser.getRemainingArgs();
    // Parse the rest, NN specific args.
    StartupOption startOpt = parseArguments(argv);
    if (startOpt == null) {
      printUsage(System.err);
      return null;
    }
    setStartupOption(conf, startOpt);
    //启动 NameNode 时的命令行选项
    switch (startOpt) {
      case FORMAT: { //HDFS 的格式化
        boolean aborted = format(conf, startOpt.getForceFormat(),
            startOpt.getInteractiveFormat());
        terminate(aborted ? 1 : 0);
        return null; // avoid javac warning
      }
      ......
      case ROLLBACK: { //回滚到原先的版本
        boolean aborted = doRollback(conf, true);
        terminate(aborted ? 1 : 0);
        return null; // avoid warning
      }
      //使安排用于 NameNode 热备份的节点复制当班 NameNode 的内容,以形成热备份
      case BOOTSTRAPSTANDBY: {
        String toolArgs[] = Arrays.copyOfRange(argv, 1, argv.length);
        int rc = BootstrapStandby.run(toolArgs, conf);
        terminate(rc);
        return null; // avoid warning
      }
      //备份
      case BACKUP:
      //生成 CHECKPOINT
      case CHECKPOINT: {
        NamenodeRole role = startOpt.toNodeRole();
        DefaultMetricsSystem.initialize(role.toString().replace(" ", ""));
        return new BackupNode(conf, role);
      }
      //试图修复损坏了的 HDFS 文件系统目录
      case RECOVER: {
        NameNode.doRecovery(startOpt, conf);
        return null;
      }
      ......
      //版本更新升级
      case UPGRADEONLY: {
        DefaultMetricsSystem.initialize("NameNode");
        new NameNode(conf);
        terminate(0);
        return null;
      }
      default: {
        DefaultMetricsSystem.initialize("NameNode");
        //没有使用任何选项,才是真正要启动 NameNode
        return new NameNode(conf);
      }
    }
  }

//构造NameNode
protected NameNode(Configuration conf, NamenodeRole role) {
    ......
    try {
      initializeGenericKeys(conf, nsId, namenodeId);
      //初始化
      initialize(conf);
    }
    this.started.set(true);
  }
  
}

//初始化系统
protected void initialize(Configuration conf) throws IOException {
	......
	//创建并启动 NameNodeHttpServer
    if (NamenodeRole.NAMENODE == role) {
      startHttpServer(conf);
    }
    //加载目录系统
    loadNamesystem(conf);
    //创建RPC服务
    rpcServer = createRpcServer(conf);
    if (NamenodeRole.NAMENODE == role) {
      httpServer.setNameNodeAddress(getNameNodeAddress());
      httpServer.setFSImage(getFSImage());
    }
    //启动各项服务
    startCommonServices(conf);
    startMetricsLogger(conf);
  }
  
 }

2. 目录系统 FSNamesystem

NameNode 的核心是 FSNamesystem ,这就是 HDFS 的目录系统。同时,从某种意义上说,FSNamesystem 也是关于集群内众多 DataNode 的记账系统。如果把那些 DataNode 看成一个个的仓库,那么 FSNamesystem 就是集中的账务管理部门。

public class FSNamesystem implements Namesystem, FSNamesystemMBean, NameNodeMXBean {

 FSDirectory dir; //目录框架
 FSImagefsImage; //文件系统镜像
 String blockPoolId;
 private final BlockManager blockManager; //块管理器
 private final CacheManager cacheManager; //缓存管理器
 private  NameNodeResourceChecker; //用于资源检查

 FSNamesystem(Configuration conf, FSImage fsImage, boolean ignoreRetryCache) {
    ......
    //作为参数传下来的一个可能有待加载内容的 FSImage 对象
    this.fsImage = fsImage;
    try {
      // 创建块管理器
      this.blockManager = new BlockManager(this, conf);
      // 统计信息
      this.datanodeStatistics = blockManager.getDatanodeManager().getDatanodeStatistics();
      //块ID管理
      this.blockIdManager = new BlockIdManager(blockManager);
      //从配置文件获取最小块容量,默认 64MB                                
      this.minBlockSize = conf.getLong(DFSConfigKeys.DFS_NAMENODE_MIN_BLOCK_SIZE_KEY,
          DFSConfigKeys.DFS_NAMENODE_MIN_BLOCK_SIZE_DEFAULT);
      //从配置文件获取单个文件的最大块数
      this.maxBlocksPerFile = conf.getLong(DFSConfigKeys.DFS_NAMENODE_MAX_BLOCKS_PER_FILE_KEY,
          DFSConfigKeys.DFS_NAMENODE_MAX_BLOCKS_PER_FILE_DEFAULT);
      //创建目录框架
      this.dir = new FSDirectory(this, conf);
      //创建缓存管理器
      this.cacheManager = new CacheManager(this, conf, blockManager);
      this.safeMode = new SafeModeInfo(conf);
      this.topConf = new TopConf(conf);
      ......
    }
  }
  
  //从磁盘装载文件系统映像
  static FSNamesystem loadFromDisk(Configuration conf) throws IOException {
    FSImage fsImage = new FSImage(conf,
        FSNamesystem.getNamespaceDirs(conf),
        FSNamesystem.getNamespaceEditsDirs(conf));
    FSNamesystem namesystem = new FSNamesystem(conf, fsImage, false);
    StartupOption startOpt = NameNode.getStartupOption(conf);
    long loadStart = monotonicNow();
    try {
      namesystem.loadFSImage(startOpt);
    }
    return namesystem;
  }
  
  //载入系统镜像
  private void loadFSImage(StartupOption startOpt) throws IOException {
    final FSImage fsImage = getFSImage();
    try {
      // 从宿主文件系统读入映像文件和 EditLog ,加以合并处理
      MetaRecoveryContext recovery = startOpt.createRecoveryContext();
      final boolean staleImage = fsImage.recoverTransitionRead(startOpt, this, recovery);
      final boolean needToSave = staleImage && !haEnabled && !isRollingUpgrade(); 
      if (needToSave) {
        //如果需要的话就将合并后的映像写回
        fsImage.saveNamespace(this);
      } 
    }
    imageLoadComplete();
  }

}

原先一个 Hadoop 集群中只能有一个 NameNode ,一个 FSNamesystem ,那就没有“ BlockPool ”这一说了。可是新版的Hadoop 开始想要支持多个 NameNode,让多个 FSNamesystem 共存于一个 Hadoop 集群,它们的数据块则可以杂处共存于数据节点上,称为“联邦( Federation )”模式。于是,就得引进“块池(BlockPool )”的概念,让不同的 NameNode ,从而不同的 FSNamesystem ,有属于自己的块池。而“块文件”的文件路径中也得带有相应的信息,带上 blockPoolId ,以区分不同的FSNamesystem ,标明一个块属于哪一个 FSNamesystem 。

3. 文件系统目录 FSDirectory

目录是个由“索引节点(indexnode )”构成的树状结构,树上的每个节点都是一个索引节点。索引节点至少有两种:一种是作为一级目录的索引节点,称为目录节点,目录节点可以出现在树上的任何位置上;另一种是作为一个文件的索引节点,称为文件节点,文件节点只能出现在树枝的末梢即被称为“叶节点”的位置上。每个节点都有个节点名,树的根是一个名为Root 的目录节点,以符号“/”表示。

public abstract class INode implements INodeAttributes, Diff.Element<byte[]> {......}
public abstract class INodeWithAdditionalFields extends INode implements LinkedElement {......}

INodeFile 在 INodeWithAdditionalFields 的 基 础 上 增 添 的 成 分 主 要 是 一 个BlockInfo 数组,文件中包含了多少个块,这个数组就相应有多大。这个数组中的每个元素都是对于一个块的描述。header 是把三种信息编码在一起而成的无符号整数。这三种信息是preferredBlockSize 、 replication 和storagePolicyID 。其中 preferredBlockSize 是文件主所希望的块的大小,默认值为 64MB ;replication 是副本的个数,默认值为 3 ; storagePolicyID 是块的存储策略编号。数组 blocks 是个 BlockInfo [],其每个元素都是一个 BlockInfo 对象,里面有 Block 的块号,也有关于各个复份存储在什么节点上的信息。

public class INodeFile extends INodeWithAdditionalFields implements INodeFileAttributes, BlockCollection {......}
public class INodeDirectory extends INodeWithAdditionalFieldsimplements INodeDirectoryAttributes {......}

FSNamesystem 内部有两大部件,一个是 FSImage ,另一个就是 FSDirectory，前者是为后者服务的,只是为了解决其持久存储的问题,FSDirectory 才是运行时的“活”的目录系统。FSNamesystem有两层，第一层是文件路径到块的映射，第二层是块到存储位置的映射。第二层是在运行时由存储数据的Node动态的上报给FSNamesystem，从而形成“活”的目录系统。

4. 文件系统映像 FsImage

FSDirectory 对象只是存在于 NameNode 所在节点机的内存中,尽管这机器不会轻易就关机断电,但这样的存在毕竟只是短暂(ephemeral )的,而不是永久的存在。然而 HDFS 文件系统中那些文件的内容却不会因此而消失,它们的存在是持续的(persistent )。我们总不能让持续存在的文件内容因为目录信息的消逝而变得不可访问,解决这个问题的唯一办法就是使目录信息也得到持久的存储。文件系统的结构信息实际上分成两个层次:第一个层次是从具体文件的路径(和具体内容在文件中的位置)到文件内容块号的映射,这一层映射是固定不变的,应该持久存储。第二层的映射则是从块号到存储地点的映射,在单机上这一层映射也是不变的,事实上关于存储地点(位置)的信息往往就编码在块号中。

public class FSImage implements Closeable {

  FSEditLogeditLog //文件系统变更日志
  NNStoragestorage //代表着存储介质,实际上是宿主文件系统中的一系列目录和文件
  protected FSImage(Configuration conf,Collection<URI> imageDirs,List<URI> editsDirs){
    this.conf = conf;
    storage = new NNStorage(conf, imageDirs, editsDirs);
    this.editLog = FSEditLog.newInstance(conf, storage, editsDirs);
    archivalManager = new NNStorageRetentionManager(conf, storage, editLog);
  }
  protected doUpgrade (FSNamesystem target ) //软件版本升级
  protected doRollback (FSNamesystemfsns ) //回滚版本升级
  protected rollEditLog () // RditLog 的滚进
  //存储
  protected saveFSImage (SaveNamespaceContext context , StorageDirectory sd , NameNodeFiled stType )
  class FSImageSaver implements Runnable {}
  protected saveNamespace (FSNamesystem source , NameNodeFile nnf , Canceler canceler)
  
}

FSImage 对象内部并没有以数据形式出现的映像,而是提供了一些生成和处理映像的方法,真正的映像则以文件的形式存在于宿主文件系统中。当创建一个FSImage 对象时,从它的构造函数可以看出同时也创建了它的三个部件,其中的两个是关键性的:一个是 NNStorage 类的对象 storage ;还有一个是 FSEditLog 类的对象 editLog 。创建这两个部件所得结果是磁盘上的目录和文件,而不仅仅是内存中的数据结构。

一个 StorageInfo 对象就是一项存储信息,里面的内容包括存储格局(layout )的版本号layoutVersion ,软件的版本升级往往就伴随着 HDFS 存储格局的改变。

public class StorageInfo {
  public int   layoutVersion;   // layout version of the storage data
  public int   namespaceID;     // id of the file system
  public String clusterID;      // id of the cluster
  public long  cTime;           // creation time of the file system state
 }

所谓一个 Storage ,实际上就是具体节点上宿主文件系统中的一个子树,该子树的根目录中有urrent 、 previous 等子目录,也可以有例如 previous. tmp 、 finalized. tmp 、 previous. checkpoint 等文件。

public abstract class Storage extends StorageInfo {
  String STORAGE _ FILE _ LOCK ="in _ use.lock" //用于加锁
  String STORAGE _ DIR _ CURRENT ="current"  //当前的文件系统结构
  String STORAGE _ DIR _ PREVIOUS="previous"  //先前的文件系统结构,以备回滚
  String STORAGE _ TMP _ REMOVED ="removed.tmp" //记载着已经删除的文件和目录
  String STORAGE _ TMP _ PREVIOUS="previous.tmp"  //记载着先前的文件和目录
  String STORAGE _ TMP _ FINALIZED="finalized.tmp"  //记载着已经封存的文件和目录
  String STORAGE _ TMP _ LAST _ CKPT="lastcheckpoint.tmp"  //最近一个 checkpoint
  String STORAGE _ PREVIOUS _ CKPT="previous.checkpoint"  //前一个 checkpoint
 
  //用作存储设备的宿主系统目录
  public static class StorageDirectory implements FormatConfirmable {}
}

NNStorage 则是对 Storage 的扩充, NNStorage 内部定义了一个枚举类型 NameNodeFile 说明了10种文件类型，内部的NameNodeDirType implements StorageDirType {} 目录的类型有 4 种。

public class NNStorage extends Storage implements Closeable, StorageErrorReporter {
}

HDFS 文件系统的 FSImage 就代表着需要加以持久存储的目录结构和第一层映射,所以这个映像并不是简单的 FSDirectory 串行化,而必须从 FSDirectory 中抽取应该持久存储的信息以形成文件系统的映像。

5. 文件系统记录 FSEditLog

Hadoop 源码中定义了一个关于文件操作的枚举类型作为操作代码, 凡是对 HDFS 文件系统的结构、内容、状态造成某种改变的操作,全都在这里了。

public enum FSEditLogOpCodes {
  // last op code in file
  OP_ADD                        ((byte)  0, AddOp.class),
  // deprecated operation
  OP_RENAME_OLD                 ((byte)  1, RenameOldOp.class),
  OP_DELETE                     ((byte)  2, DeleteOp.class),
  OP_MKDIR                      ((byte)  3, MkdirOp.class),
  OP_SET_REPLICATION            ((byte)  4, SetReplicationOp.class),
  @Deprecated OP_DATANODE_ADD   ((byte)  5), // obsolete
  @Deprecated OP_DATANODE_REMOVE((byte)  6), // obsolete
  OP_SET_PERMISSIONS            ((byte)  7, SetPermissionsOp.class),
  // Note that the current range of the valid OP code is 0~127
  OP_INVALID                    ((byte) -1);
  ......
  private final byte opCode;
  private final Class<? extends FSEditLogOp> opClass;
}

Hadoop 定义了一个抽象类 FSEditLogOp,作为所有操作记录的基础,然后针对不同的操作可以扩充这个抽象类,加上不同的信息。

public abstract class FSEditLogOp {
  public final FSEditLogOpCodes opCode;
  long txid;
  byte[] rpcClientId;
  int rpcCallId;
  
  static class AddBlockOp extends FSEditLogOp {
    private String path;
    private Block penultimateBlock;
    private Block lastBlock;
    ......
  }
  
}

FSEditLog 类对象属于 FSImage ,是 FSImage 内部的一个成分。

public class FSEditLog implements LogsPurgeable {

  synchronized void openForWrite(int layoutVersion) throws IOException {
     long segmentTxId = getLastWrittenTxId() + 1;
     List<EditLogInputStream> streams = new ArrayList<EditLogInputStream>();
     journalSet.selectInputStreams(streams, segmentTxId, true);
     startLogSegment(segmentTxId, true, layoutVersion);
     assert state == State.IN_SEGMENT : "Bad state: " + state;
  }
  private long beginTransaction () //开始一次日志写入
  private long endTransaction (longstart )//结束一次日志写入
  private long logSync () //对缓冲着的日志进行同步
  private void startLogSegment (final long segmentTxId , boolean writeHeaderTxn ) //打开一个日志段
  private long endCurrentLogSegment (boolean writeEndTxn )//关闭当前日志段
  private long purgeLogsOlderThan (final long minTxIdToKeep ) //冲洗掉日志中过老的记录
  private long getJournalClass (Configurationconf , StringuriScheme ) //获取用作 Journal 的类
  private long createJournal (URIuri ) //创建 Journal 对象

}

把 FSEditLog 持久存储下来,就称为 Journal ,在内存中则以 Journal 对象为代表。一个Journal 对象代表着 FSEditLog 的一份持久存储。之所以要有 EditLog ,就是为了可以在 FSImage 上重演。这样,对于 FSImage 就可以过一段(不短的)时间才做一个硬拷贝,因为那个开销太大;平时对 FSImage 的操作则随时记录在 FSEditLog 的硬拷贝中; 万一发生事故,两个硬拷贝都还在,把记录在 EditLog 中的操作在(已经过时的)文件系统映像上重演一遍,就恢复过来了。