HDFS追本溯源:HDFS操作的逻辑流程与源码解析

  1. 云栖社区>
  2. 博客>
  3. 正文

HDFS追本溯源:HDFS操作的逻辑流程与源码解析

anzhsoft 2014-04-13 19:37:00 浏览976
展开阅读全文

本文主要介绍5个典型的HDFS流程,这些流程充分体现了HDFS实体间IPC接口和stream接口之间的配合。

1. Client和NN

      Client到NN有大量的元数据操作,比如修改文件名,在给定目录下创建一个子目录,这些操作一般只涉及Client和NN的交互,通过IPC调用ClientProtocol进行。创建子目录的逻辑流程如下图:

      从图中可见,创建子目录这种操作并没有涉及DN。因为元数据会被NN持久化到edits中,因此在持久化结束之后,这个调用就会被成功返回。复习一下:NN维护了HDFS的文件系统目录树和文件与数据块的对应关系,和数据块与DN的对应关系。因此,创建目录仅仅是在NN上也就很容易理解了。

     一些更为复杂的操作,如使用

DistributedFileSystem.setReplication()
来增加文件的副本数,再如通过

DistributedFileSystem.delete()
来删除HDFS上的文件,都需要DN配合执行一些动作。其中DistributedFileSystem源码在hadoop-hdfs-project\hadoop-hdfs\src\main\java\org\apache\hadoop\hdfs\DistributedFileSystem.java

/****************************************************************
 * Implementation of the abstract FileSystem for the DFS system.
 * This object is the way end-user code interacts with a Hadoop
 * DistributedFileSystem.
 *
 *****************************************************************/
@InterfaceAudience.LimitedPrivate({ "MapReduce", "HBase" })
@InterfaceStability.Unstable
public class DistributedFileSystem extends FileSystem {
  private Path workingDir;
  private URI uri;

  DFSClient dfs;
  private boolean verifyChecksum = true;
  
  static{
    HdfsConfiguration.init();
  }

    以客户端删除HDFS文件为例,操作在NN上执行完成后,DN存放的文件内容的数据块也必须删除。但是,NN在执行delete()方法时,它只标记需要删除的数据块(当然,delete的操作日志也会被持久化),而不会主动联系DN去立即删除这些数据。当保存着这些数据块的DN在向NN发送心跳时,NN会通过心跳应答携带DatanodeCommand命令来通知DN删除数据。也就是说,被删除的数据块在Client接到成功的响应后,会在一段时间后才能真正删除,NN和DN永远只维护简单的主从关系。NN永远不会主动发起向DN的调用。NN只能通过DN心跳应答中携带DatanodeCommand的指令对DN进行管理。


2. Client读文件

     使用Java API读取文件的源码如下:

FileSystem hdfs = FileSystem.get(new Configuration());
Path path = new Path("/testfile");// reading
FSDataInputStream dis = hdfs.open(path);
byte[] writeBuf = new byte[1024];
int len = dis.read(writeBuf);
System.out.println(new String(writeBuf, 0, len, "UTF-8"));
dis.close();

hdfs.close();
    下图显示了HDFS在读取文件时,Client,NN和DN发生的事件和这些事件的顺序:

步骤1的源码:

public FSDataInputStream open(Path f, final int bufferSize)
      throws IOException {
    statistics.incrementReadOps(1);
    Path absF = fixRelativePart(f);
    return new FileSystemLinkResolver<FSDataInputStream>() {
      @Override
      public FSDataInputStream doCall(final Path p)
          throws IOException, UnresolvedLinkException {
        return new HdfsDataInputStream(
            dfs.open(getPathName(p), bufferSize, verifyChecksum));
      }
      @Override
      public FSDataInputStream next(final FileSystem fs, final Path p)
          throws IOException {
        return fs.open(p, bufferSize);
      }
    }.resolve(this, absF);
  }
可见open返回的是HdfsDataInputStream。dfs为hadoop-hdfs-project\hadoop-hdfs\src\main\java\org\apache\hadoop\hdfs\DFSClient.java。HdfsDataInputStream继承自FSDataInputStream。构造是并没有额外的处理。

public class HdfsDataInputStream extends FSDataInputStream {
  public HdfsDataInputStream(DFSInputStream in) throws IOException {
    super(in);
  }
FSDataInputStream继承自DFSInputStream。
DFSInputStream(DFSClient dfsClient, String src, int buffersize, boolean verifyChecksum
                 ) throws IOException, UnresolvedLinkException {
    this.dfsClient = dfsClient;
    this.verifyChecksum = verifyChecksum;
    this.buffersize = buffersize;
    this.src = src;
    this.cachingStrategy =
        dfsClient.getDefaultReadCachingStrategy();
    openInfo();
  }

  /**
   * Grab the open-file info from namenode
   */
  synchronized void openInfo() throws IOException, UnresolvedLinkException {
    lastBlockBeingWrittenLength = fetchLocatedBlocksAndGetLastBlockLength();
    int retriesForLastBlockLength = dfsClient.getConf().retryTimesForGetLastBlockLength;
    while (retriesForLastBlockLength > 0) {
      // Getting last block length as -1 is a special case. When cluster
      // restarts, DNs may not report immediately. At this time partial block
      // locations will not be available with NN for getting the length. Lets
      // retry for 3 times to get the length.
      if (lastBlockBeingWrittenLength == -1) {
        DFSClient.LOG.warn("Last block locations not available. "
            + "Datanodes might not have reported blocks completely."
            + " Will retry for " + retriesForLastBlockLength + " times");
        waitFor(dfsClient.getConf().retryIntervalForGetLastBlockLength);
        lastBlockBeingWrittenLength = fetchLocatedBlocksAndGetLastBlockLength();
      } else {
        break;
      }
      retriesForLastBlockLength--;
    }
    if (retriesForLastBlockLength == 0) {
      throw new IOException("Could not obtain the last block locations.");
    }
  }

fetchLocatedBlocksAndGetLastBlockLength通过调用getLocatedBlocks实现了示意图中的步骤二:

  private long fetchLocatedBlocksAndGetLastBlockLength() throws IOException {
    final LocatedBlocks newInfo = dfsClient.getLocatedBlocks(src, 0);
    if (DFSClient.LOG.isDebugEnabled()) {
      DFSClient.LOG.debug("newInfo = " + newInfo);
    }
    if (newInfo == null) {
      throw new IOException("Cannot open filename " + src);
    }

    if (locatedBlocks != null) {
      Iterator<LocatedBlock> oldIter = locatedBlocks.getLocatedBlocks().iterator();
      Iterator<LocatedBlock> newIter = newInfo.getLocatedBlocks().iterator();
      while (oldIter.hasNext() && newIter.hasNext()) {
        if (! oldIter.next().getBlock().equals(newIter.next().getBlock())) {
          throw new IOException("Blocklist for " + src + " has changed!");
        }
      }
    }
    locatedBlocks = newInfo;
    long lastBlockBeingWrittenLength = 0;
    if (!locatedBlocks.isLastBlockComplete()) {
      final LocatedBlock last = locatedBlocks.getLastLocatedBlock();
      if (last != null) {
        if (last.getLocations().length == 0) {
          if (last.getBlockSize() == 0) {
            // if the length is zero, then no data has been written to
            // datanode. So no need to wait for the locations.
            return 0;
          }
          return -1;
        }
        final long len = readBlockLength(last);
        last.getBlock().setNumBytes(len);
        lastBlockBeingWrittenLength = len; 
      }
    }

    currentNode = null;
    return lastBlockBeingWrittenLength;
  }
你可能会说步骤二调用的是getBlockLocations。看以下的代码:

@VisibleForTesting
  public LocatedBlocks getLocatedBlocks(String src, long start, long length)
      throws IOException {
    return callGetBlockLocations(namenode, src, start, length);
  }

  /**
   * @see ClientProtocol#getBlockLocations(String, long, long)
   */
  static LocatedBlocks callGetBlockLocations(ClientProtocol namenode,
      String src, long start, long length) 
      throws IOException {
    try {
      return namenode.getBlockLocations(src, start, length);
    } catch(RemoteException re) {
      throw re.unwrapRemoteException(AccessControlException.class,
                                     FileNotFoundException.class,
                                     UnresolvedPathException.class);
    }
  }
  

      然后就可以开始读文件的数据了。通过NameNode.getBlockLocations的远程调用接口获得了文件开始部分的数据块的保存位置。对于文件中的每个块,NN返回保存着该副本的DN的地址。注意,这些DN根据它们与Client的距离进行了简单的排序(利用了网络的拓扑信息)。

      Client调用HdfsDataInputStream的read方法读取文件数据时,DFSInputStream对象会通过和DN间的读数据stream接口,和最近的DN建立连接。Client反复调用read方法,数据会通过DN和Client的连接上的数据包返回Client。当到达块的末端时,DFSInputStream会关闭和DN的连接。并通过getBlockLocations()远程方法获得保存着下一个数据块的DN信息,严格来说,在对象没有缓存该数据块的位置时,才会使用这个远程方法。这就是上图中的步骤五。然后重复上述过程。

      另外,由于NameNode.getBlockLocations()不会一次返回文件的所有的数据块信息,DFSInputStream可能需要多次调用该远程方法,检索下一组数据块的位置信息。对于使用者来说,它读取的是一个连续的数据流,上面所讲的联系不同的DN,多次定位数据块的过程,都是透明的。当使用者完成数据读取任务后,通过FSDataInputStream.close()关系数据流。即图中的步骤六。

     如果DN发生了错误,如节点停机或者网络出现故障,那么Client会尝试连接下一个Block的位置。同时它会记住出现故障的那个DN,不会再进行徒劳的尝试。在数据的应答中,不单包含了数据,还包含了数据的校验和,Client会检查数据的一致性,如果发现了校验错误,它会将这个信息报告给NN;同时,尝试从别的DN读取另外一个副本的内容。由Client在读取数据时进行数据完整性检查,可以降低DN的负载,均衡各个节点的计算能力。

     这样的设计其实可以给我们一个很好的设计大型分布式系统的例子。通过一些有效的设计,将计算和网络等分散到各个节点上,这样可以最大程度的保证scalability。

3. Client写文件

  即使不考虑出现错误的情况,写文件也是HDFS最复杂的流程。本节通过创建一个新文件并向文件写入数据,结束后关闭这个文件为例,分析文件写入时各个节点之间的配合。


      Client调用DistributedFileSystem.create()创建文件(上图中的步骤一),这时DistributedFileSystem创建了DFSOutputStream,并由RPC,让NN执行同名的方法,在文件系统的命名空间创建一个新文件。NN创建新文件时,需要执行检查,包括NN是否处理正常工作状态,被创建的文件不存在,Client是否有在父目录中创建文件的权限等。通过检查后,NN会构建一个新文件,记录创建操作到编辑日志edits中。RPC结束后,DistributedFileSystem将该DFSOutputStream对象包装到FSDataOutputStream实例中,返回Client。


 @Override
  public FSDataOutputStream create(final Path f, final FsPermission permission,
    final EnumSet<CreateFlag> cflags, final int bufferSize,
    final short replication, final long blockSize, final Progressable progress,
    final ChecksumOpt checksumOpt) throws IOException {
    statistics.incrementWriteOps(1);
    Path absF = fixRelativePart(f);
    return new FileSystemLinkResolver<FSDataOutputStream>() {
      @Override
      public FSDataOutputStream doCall(final Path p)
          throws IOException, UnresolvedLinkException {
        return new HdfsDataOutputStream(dfs.create(getPathName(p), permission,
            cflags, replication, blockSize, progress, bufferSize, checksumOpt),
            statistics);
      }
      @Override
      public FSDataOutputStream next(final FileSystem fs, final Path p)
          throws IOException {
        return fs.create(p, permission, cflags, bufferSize,
            replication, blockSize, progress, checksumOpt);
      }
    }.resolve(this, absF);
  }

关键的调用点有DFSClient.create:

 /**
   * Same as {@link #create(String, FsPermission, EnumSet, boolean, short, long,
   * Progressable, int, ChecksumOpt)} with the addition of favoredNodes that is
   * a hint to where the namenode should place the file blocks.
   * The favored nodes hint is not persisted in HDFS. Hence it may be honored
   * at the creation time only. HDFS could move the blocks during balancing or
   * replication, to move the blocks from favored nodes. A value of null means
   * no favored nodes for this create
   */
  public DFSOutputStream create(String src, 
                             FsPermission permission,
                             EnumSet<CreateFlag> flag, 
                             boolean createParent,
                             short replication,
                             long blockSize,
                             Progressable progress,
                             int buffersize,
                             ChecksumOpt checksumOpt,
                             InetSocketAddress[] favoredNodes) throws IOException {
    checkOpen();
    if (permission == null) {
      permission = FsPermission.getFileDefault();
    }
    FsPermission masked = permission.applyUMask(dfsClientConf.uMask);
    if(LOG.isDebugEnabled()) {
      LOG.debug(src + ": masked=" + masked);
    }
    String[] favoredNodeStrs = null;
    if (favoredNodes != null) {
      favoredNodeStrs = new String[favoredNodes.length];
      for (int i = 0; i < favoredNodes.length; i++) {
        favoredNodeStrs[i] = 
            favoredNodes[i].getHostName() + ":" 
                         + favoredNodes[i].getPort();
      }
    }
    final DFSOutputStream result = DFSOutputStream.newStreamForCreate(this,
        src, masked, flag, createParent, replication, blockSize, progress,
        buffersize, dfsClientConf.createChecksum(checksumOpt), favoredNodeStrs);
    beginFileLease(src, result);
    return result;
  }

在步骤三Client写入数据时,由于create()调用创建了一个空文件,所以,DFSOutputStream实例首先需要想NN申请数据块,addBlock()方法成功执行后,返回一个LocatedBlock对象。该对象包含了新数据块的数据块标识和版本好,同时,它的成员变量locs提供了数据流管道的信息,通过上述信息,DFSOutputStream就可以和DN连接,通过些数据接口建立数据流管道。Client写入FSDataOutputStream流中的数据,被分成一个一个的文件包,放入DFSOutputStream对象的内部队列。该队列中的文件包最后打包成数据包,发往数据流管道,流经管道上的各个DN,并持久化,确认包逆流而上,从数据流管道依次发往Client,当Client收到应答时,它将对应的包从内部队列删除。

public class LocatedBlock {

  private ExtendedBlock b;
  private long offset;  // offset of the first byte of the block in the file
  private DatanodeInfo[] locs;
  /** Storage ID for each replica */
  private String[] storageIDs;
  // Storage type for each replica, if reported.
  private StorageType[] storageTypes;
  // corrupt flag is true if all of the replicas of a block are corrupt.
  // else false. If block has few corrupt replicas, they are filtered and 
  // their locations are not part of this object
  private boolean corrupt;
  private Token<BlockTokenIdentifier> blockToken = new Token<BlockTokenIdentifier>();
  /**
   * List of cached datanode locations
   */
  private DatanodeInfo[] cachedLocs;

  // Used when there are no locations
  private static final DatanodeInfo[] EMPTY_LOCS = new DatanodeInfo[0];

    DFSOutputStream在写完一个数据块后,数据流管道上的节点,会通过和NN的DatanodeProtocol远程接口的blockReceived()方法,向NN提交数据块。如果数据队列还有等到输出的数据,DFSOutputStream会再次调用addBlock(),为文件添加新的数据块。

     Client完成数据的写入后,会调用close()方法关闭流,关闭流意味着Client不会再向流中写入数据。所以,当DFSOutputStream数据队列的文件包都收到应答后,就可以使用ClientProtocol.complete()方法通知NN关闭文件,完成一次正常的文件写入。

     如果在文件写入期间DN发生故障,则会执行下面的操作(注意,这些操作对于写入数据的Client是透明的):

  1. 数据流管道会被关闭,已经发送到管道但是还没有收到确认的文件包,会被重新添加到DFSOutputStream的输出队列,这样就保证了无路数据流管道的哪个DN发生故障,都不会丢失数据。当前正常工作的DN的数据块会被赋予一个新的版本号,并通知NN。这样,失败的DN在从故障恢复过来以后,上面只有部分数据的Block会因为版本号和NN保存的版本号不匹配而被删除。
  2. 在数据流管道中删除错误的DN并建立新的管道,继续写数据到正常工作的DN。
  3. 文件关闭后,NN会发现该Block的副本数没有达到要求,会选择一个新的DN并复制Block,创建新的副本。DN的故障只会影响一个Block的写操作,后续Block的写入不会受到影响。


4. DataNode的启动与心跳机制

      本节讨论DN的启动及其与NN之间的交互。包括DN从启动到进入正常工作状态的注册,Block上报,以及正常工作过程中的心跳等与NN相关的远程调用。这部分虽然只涉及DatanodeProtocol的接口,但是有助于进一步理解DN与NN的关系。

      正常启动的DN或者为升级而启动的DN,都会向NN发送远程调用versionRequest(),进行必要的版本检查。这里的版本检查,只涉及构建版本号,保证它们间的HDFS版本是一致的。


     在版本检查结束后,DN会接着通过远程调用register(),向NN注册。DatanodeProtocol.register()的主要工作也是检查,确认该DN是NN所管理集群的成员。也就是说,用户不能把某一个集群中的某个node直接注册到另外一个集群中去,保证了整个系统的数据一致性。

     注册成功后,DN会将它所管理的所有Block的信息,通过blockRequest()方法上报到NN(步骤三),以帮助NN建立HDFS文件数据块到DN的映射关系。在此后,DN才正式的提供服务。

 由于NN和DN是简单的主从关系,DN需要每隔一段时间发送心跳到NN(步骤四和步骤五)。如果NN长时间收不到DN的心跳,它会认为DN已经失效。如果NN需要一些DN需要配合的动作,则会通过sendHeartbeat()的方法返回。该返回值是一个DatanodeCommand数组,它是NN的指令。

     应该说,DN和NN的交互逻辑非常简单。大部分是通过DN到NN的心跳来完成的。但是考虑到一定规模的HDFS集群,一个NN会管理上千个DN,这样的设计也就非常自然了。


5. SNN节点的元数据合并

      当Client对HDFS的文件目录进行修改时,NN都会在edits中留下记录,以保证在系统出现问题时,通过日志可以进行恢复。

    fsimage是某一个时刻的检查点(checkpoint)。由于fsimage很大,因此不会在每次的元数据修改都写入到它里边,而只是存在到edits中。在系统启动时,会首先状态最近时刻的fsimage,然后在通过edits,恢复系统的最新状态。

    当时如果edits太大,那么节点启动时将用很长的时间来执行日志的每一个操作,使得系统恢复最近的状态。在启动恢复的这段时间,服务是不可用的。为了避免edits多大,增加集群的可用时间,HDFS引入了第二名字节点,即SNN(Secondary NameNode)。SNN不是NN的standby,它只是辅助NN完成合并(merge)fsimage和edits。过程涉及NamenodeProtocol和NN与SNN之间的流式接口。

    该过程由SNN发起,首先通过远程方法NamenodeProtocol.getEditLogSize()获得NN上edits的大小。如果日志很小,SNN就会在指定的时间后重新检查。否则,继续通过远程接口rollEditLog(),启动一次检查点的过程。这时,NN需要创建一个新的编辑日志edits.new,后续对元数据的改动,都会记录到这个新日志中。而原有的fsimage和edits,会由SNN通过HTTP下载到本地(步骤三和步骤四),在内存中进行merge。合并的结果就是fsimage.ckpt。然后SNN通过HTTP接口通知NN fsimage已经准备好。NN会通过HTTP get获取merge好的fsimage。在NN下载完成后,SNN会通过NamenodeProtocol.rollFsImage(),完成这次检查点。NN在处理这个远程方法时,会用fsimage.ckpt 覆盖原来的fsimage,并且将新的edits.new改名为edit。


尊重原创,未经允许不得转载:http://www.anzhan.me/index.php/archives/233

参考资料:

1. Hadoop技术内幕-深入解析HADOOP COMMON和HDFS架构设计与实现原理

网友评论

登录后评论
0/500
评论
anzhsoft
+ 关注