HDFS

本文主要介绍hadoop2.x版本，分析代码主要位于hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server下namenode、datanode与protocol。

目录：

1. HDFS介绍
2. HDFS架构
3. HDFS读写过程
4. HDFS读写源码解析

HDFS介绍

参考：https://hadoop.apache.org/doc...

hdfs是适用于通用硬件上的分布式文件系统，具有高容错性与高吞吐量等特点，其设计理念如下：

1. 硬件错误：硬件错误是常态而不是异常。HDFS可能由成百上千的服务器所构成，每个服务器上存储着文件系统的部分数据。我们面对的现实是构成系统的组件数目是巨大的，而且任一组件都有可能失效，这意味着总是有一部分HDFS的组件是不工作的。因此错误检测和快速、自动的恢复是HDFS最核心的架构目标。
2. 流式数据访问：运行在HDFS上的应用和普通的应用不同，需要流式访问它们的数据集。HDFS的设计中更多的考虑到了数据批处理，而不是用户交互处理。比之数据访问的低延迟问题，更关键的在于数据访问的高吞吐量。POSIX标准设置的很多硬性约束对HDFS应用系统不是必需的。为了提高数据的吞吐量，在一些关键方面对POSIX的语义做了一些修改。
3. 大规模数据集：运行在HDFS上展到数百个节点。一个单一的HDFS实例应该能支撑数以千万计的文件。
4. 简单的一致性模型：HDFS应用需要一个“一次写入多次读取”的文件访问模型。一个文件经过创建、写入和关闭之后就不需要改变。这一假设简化了数据一致性问题，并且使高吞吐量的数据访问成为可能。Map/Reduce应用或者网络爬虫应用都非常适合这个模型。目前还有计划在将来扩充这个模型，使之支持文件的附加写操作。
5. 移动计算比移动数据更划算:一个应用请求的计算，离它操作的数据越近就越高效，在数据达到海量级别的时候更是如此。因为这样就能降低网络阻塞的影响，提高系统数据的吞吐量。将计算移动到数据附近，比之将数据移动到应用所在显然更好。HDFS为应用提供了将它们自己移动到数据附近的接口。
6. 异构软硬件平台间的可移植性:HDFS在设计的时候就考虑到平台的可移植性。这种特性方便了HDFS作为大规模数据应用平台的推广。

HDFS架构

来源：https://juejin.im/post/5a656c...

hdfs集群包含两类角色，一种称作namenode，一种称作datanode，namenode用于记录数据的元数据，元数据主要包含两类文件fsimage与editlog;

• fsimage ：保存了最新的元数据检查点，包含了整个HDFS文件系统的所有目录和文件的信息。对于文件来说包括了数据块描述信息、修改时间、访问时间等；对于目录来说包括修改时间、访问权限控制信息(目录所属用户，所在组)等。简单的说，Fsimage就是在某一时刻，整个hdfs 的快照，就是这个时刻hdfs上所有的文件块和目录，分别的状态，位于哪些个datanode，各自的权限，各自的副本个数等。注意：Block的位置信息不会保存到fsimage，Block保存在哪个DataNode（由DataNode启动时上报）。
• editlog ：主要是在NameNode已经启动情况下对HDFS进行的各种更新操作进行记录，HDFS客户端执行所有的写操作都会被记录到editlog中。

datanode保存了实际数据，数据分为两类,一类为block文件是数据本身，另一类为block.meta文件包含CRC校验和等block元信息。
相比与hadoop1.x版本2.x版本多了HA与Federation概念，主要用于解决1.x版本时namenode单点故障问题，原来的namenode以namespace服务的形式对外服务，包含active namenode与standby namenode，active namenode为正在作用的namenode，standby namenode为备用。Federation则将多个集群的视图合并为一个，方便用户的使用。

HDFS读写过程

参考：https://blog.csdn.net/gaijian...
https://www.cnblogs.com/andy6...

读取过程


HDFS Client通过请求namenode获得数据的元数据，如文件位置、文件大小、修改时间、访问权限等，再根据数据块所在数据节点的距离进行读取，从近到远依次读取。

写入过程

HDFS Client向namenode发出写入请求，namenode经过权限等检查后为写入创建位置等元数据并返回给Client，Client根据信息向datanode写入数据，数据经过分块后同时向不同的datanode写入，相同数据块的复制则是在不同的datanode建进行。

HDFS写入源码解析

源码版本为2.7.3

由HDFS的写入过程可知，写入过程分为：1.client向namenode发出写请求，namenode创建目录并将相关信息返回给client；2.client携带信息向datanode写入数据；3.datanode之间进行备份。下面依次分析这一过程：

namenode创建目录

参考：https://juejin.im/post/5aa5dc... ,版本上略有差异。

namenode创建目录的流程：1.客户端通过ClientProtocol协议向RpcServer发起创建目录的RPC请求；2.RpcServer调用FSNamesystem进行相关检查；3.FSNamesystem调用FSDirMkdirOp中的相关方法在目录树中创建目标目录，并通过日志系统备份文件系统的修改；4.RpcServer将RPC响应返回给客户端。

RpcServer

//ClientProtocol.mkdirs();
  @Idempotent
  boolean mkdirs(String src, FsPermission masked, boolean createParent)
      throws IOException;

//NamenodeProtocols extends ClientProtocol
//NameNodeRpcServer implements NamenodeProtocols
//NameNodeRpcServer.mkdirs()
 protected final FSNamesystem namesystem;

 @Override 
  public boolean mkdirs(String src, FsPermission masked, boolean createParent)
      throws IOException {
    checkNNStartup();
    if(stateChangeLog.isDebugEnabled()) {
      stateChangeLog.debug("*DIR* NameNode.mkdirs: " + src);
    }
    //NameNodeRpcServer仅做一些路径的简单检查
    if (!checkPathLength(src)) {
      throw new IOException("mkdirs: Pathname too long.  Limit " 
                            + MAX_PATH_LENGTH + " characters, " + MAX_PATH_DEPTH + " levels.");
    }
    return namesystem.mkdirs(src,
        new PermissionStatus(getRemoteUser().getShortUserName(),
            null, masked), createParent);
  }

FSNamesystem

//FSNamesystem.mkdirs()进行了更细致的权限检查，并调用FSDirMkdirOp.mkdirs进行目录创建
/**
   * Create all the necessary directories
   */
  boolean mkdirs(String src, PermissionStatus permissions,
      boolean createParent) throws IOException {
    final String operationName = "mkdirs";
    FileStatus auditStat = null;
    checkOperation(OperationCategory.WRITE);
    final FSPermissionChecker pc = getPermissionChecker();
    writeLock();
    try {
      checkOperation(OperationCategory.WRITE);
      checkNameNodeSafeMode("Cannot create directory " + src);
      auditStat = FSDirMkdirOp.mkdirs(this, pc, src, permissions,
          createParent);
    } catch (AccessControlException e) {
      logAuditEvent(false, operationName, src);
      throw e;
    } finally {
      writeUnlock(operationName);
    }
    getEditLog().logSync();
    logAuditEvent(true, operationName, src, null, auditStat);
    return true;
  }

FSDirMkdirOp

static FileStatus mkdirs(FSNamesystem fsn, FSPermissionChecker pc, String src,
      PermissionStatus permissions, boolean createParent) throws IOException {
    FSDirectory fsd = fsn.getFSDirectory();
    if(NameNode.stateChangeLog.isDebugEnabled()) {
      NameNode.stateChangeLog.debug("DIR* NameSystem.mkdirs: " + src);
    }
    fsd.writeLock();
    try {
      INodesInPath iip = fsd.resolvePath(pc, src, DirOp.CREATE);

      final INode lastINode = iip.getLastINode();
      if (lastINode != null && lastINode.isFile()) {
        throw new FileAlreadyExistsException("Path is not a directory: " + src);
      }

      if (lastINode == null) {
        if (fsd.isPermissionEnabled()) {
          fsd.checkAncestorAccess(pc, iip, FsAction.WRITE);
        }

        if (!createParent) {
          fsd.verifyParentDir(iip);
        }

        // validate that we have enough inodes. This is, at best, a
        // heuristic because the mkdirs() operation might need to
        // create multiple inodes.
        fsn.checkFsObjectLimit();

        // Ensure that the user can traversal the path by adding implicit
        // u+wx permission to all ancestor directories.
        INodesInPath existing =
            createParentDirectories(fsd, iip, permissions, false);
        if (existing != null) {
          existing = createSingleDirectory(
              fsd, existing, iip.getLastLocalName(), permissions);
        }
        if (existing == null) {
          throw new IOException("Failed to create directory: " + src);
        }
        iip = existing;
      }
      return fsd.getAuditFileInfo(iip);
    } finally {
      fsd.writeUnlock();
    }
  }

datanode数据写入

参考：https://juejin.im/post/5a77b9...

DataTransferProtocol用于整个管道中的客户端、数据节点间的流式通信，其中，DataTransferProtocol.writeBlock()负责完成写数据块的工作,datanode中用于存储的主要有线程DataXceiverServer、DataXceiver与BPServiceActor,DataXceiver主要用于数据的接收。

DataTransferProtocol.writeBlock()

void writeBlock(final ExtendedBlock blk,
      final StorageType storageType,
      final Token<BlockTokenIdentifier> blockToken,
      final String clientName,
      final DatanodeInfo[] targets,
      final StorageType[] targetStorageTypes,
      final DatanodeInfo source,
      final BlockConstructionStage stage,
      final int pipelineSize,
      final long minBytesRcvd,
      final long maxBytesRcvd,
      final long latestGenerationStamp,
      final DataChecksum requestedChecksum,
      final CachingStrategy cachingStrategy,
      final boolean allowLazyPersist,
      final boolean pinning,
      final boolean[] targetPinnings,
      final String storageID,
      final String[] targetStorageIDs) throws IOException;

DataXceiverServer
DataTransferProtocol.writeBlock()由DataXceiver实现，而DataXceiver线程由DataXceiverServer线程创建。

 @Override
  public void run() {
    Peer peer = null;
    while (datanode.shouldRun && !datanode.shutdownForUpgrade) {
      try {
        peer = peerServer.accept();

        // Make sure the xceiver count is not exceeded
        int curXceiverCount = datanode.getXceiverCount();
        if (curXceiverCount > maxXceiverCount) {
          throw new IOException("Xceiver count " + curXceiverCount
              + " exceeds the limit of concurrent xcievers: "
              + maxXceiverCount);
        }
        //创建DataXceiver进程
        new Daemon(datanode.threadGroup,
            DataXceiver.create(peer, datanode, this))
            .start();
      } catch (SocketTimeoutException ignored) {
        // wake up to see if should continue to run
      } catch (AsynchronousCloseException ace) {
        // another thread closed our listener socket - that's expected during shutdown,
        // but not in other circumstances
        if (datanode.shouldRun && !datanode.shutdownForUpgrade) {
          LOG.warn(datanode.getDisplayName() + ":DataXceiverServer: ", ace);
        }
      } catch (IOException ie) {
        IOUtils.cleanup(null, peer);
        LOG.warn(datanode.getDisplayName() + ":DataXceiverServer: ", ie);
      } catch (OutOfMemoryError ie) {
        IOUtils.cleanup(null, peer);
        // DataNode can run out of memory if there is too many transfers.
        // Log the event, Sleep for 30 seconds, other transfers may complete by
        // then.
        LOG.error("DataNode is out of memory. Will retry in 30 seconds.", ie);
        try {
          Thread.sleep(30 * 1000);
        } catch (InterruptedException e) {
          // ignore
        }
      } catch (Throwable te) {
        LOG.error(datanode.getDisplayName()
            + ":DataXceiverServer: Exiting due to: ", te);
        datanode.shouldRun = false;
      }
    }

    // Close the server to stop reception of more requests.
    try {
      peerServer.close();
      closed = true;
    } catch (IOException ie) {
      LOG.warn(datanode.getDisplayName()
          + " :DataXceiverServer: close exception", ie);
    }

    // if in restart prep stage, notify peers before closing them.
    if (datanode.shutdownForUpgrade) {
      restartNotifyPeers();
      // Each thread needs some time to process it. If a thread needs
      // to send an OOB message to the client, but blocked on network for
      // long time, we need to force its termination.
      LOG.info("Shutting down DataXceiverServer before restart");
      // Allow roughly up to 2 seconds.
      for (int i = 0; getNumPeers() > 0 && i < 10; i++) {
        try {
          Thread.sleep(200);
        } catch (InterruptedException e) {
          // ignore
        }
      }
    }
    // Close all peers.
    closeAllPeers();
  }

DataXceiver
DataXceiver.write主要分为两个过程，准备接收数据与接收数据。

//准备数据接收由BlockReceiver的构造函数完成，主要进行参数检查与设置以及创建临时文件
 BlockReceiver(final ExtendedBlock block, final StorageType storageType,
      final DataInputStream in,
      final String inAddr, final String myAddr,
      final BlockConstructionStage stage, 
      final long newGs, final long minBytesRcvd, final long maxBytesRcvd, 
      final String clientname, final DatanodeInfo srcDataNode,
      final DataNode datanode, DataChecksum requestedChecksum,
      CachingStrategy cachingStrategy,
      final boolean allowLazyPersist,
      final boolean pinning,
      final String storageId)

//接收数据由BlockReceiver.receiveBlock()完成，主要完成同步接收packet，写block文件和meta文件，发送ack以及数据复制
void receiveBlock(
      DataOutputStream mirrOut, // output to next datanode
      DataInputStream mirrIn,   // input from next datanode
      DataOutputStream replyOut,  // output to previous datanode
      String mirrAddr, DataTransferThrottler throttlerArg,
      DatanodeInfo[] downstreams,
      boolean isReplaceBlock)