Hadoop二次开发必备，Hadoop源码分析（十一）

万事俱备，我们可以来分析NameNode上的流程啦。





首先我们来看NameNode上实现的ClientProtocol，客户端通过这个接口，可以对目录树进行操作，打开/关闭文件等。getBlockLocations用于确定文件内容的位置，它的输入参数为：文件名，偏移量，长度，返回值是一个LocatedBlocks对象（如下图），它携带的信息很多，大部分字段我们以前都讨论过。




getBlockLocations直接调用NameSystem的同名方法。NameSystem中这样的方法首先会检查权限和对参数进行检查（如偏移量和长度要大于0），然后再调用实际的方法。找LocatedBlocks先找src对应的INode，然后通过INode的getBlocks方法，可以拿到该节点的Block列表，如果返回为空，表明该INode不是文件，返回null；如果Block列表长度为0，以空的Block数组构造返回的LocatedBlocks。
如果Block数组不为空，则通过请求的偏移量和长度，就可以把这个区间涉及的Block找出来，对于每一个block，执行：

l          通过BlocksMap我们可以找到它存在于几个DataNode上（BlocksMap.numNodes方法）；

l          计算包含该数据块但数据块是坏的DataNode的数目（通过NameSystem.countNodes方法，间接访问CorruptReplicasMap中的信息）；

l          计算坏数据块的数目（CorruptReplicasMap.numCorruptReplicas方法，应该和上面的数相等）；

l          通过上面的计算，我们得到现在还OK的数据块数目；

l          从BlocksMap中找出所有OK的数据块对应的DatanodeDescriptor（DatanodeInfo的父类）；

l          创建对应的LocatedBlock。

收集到每个数据块的LocatedBlock信息后，很自然就能构造LocatedBlocks对象。getBlockLocations其实只是一个读的方法，请求到了NameNode以后只需要查表就行了。
create方法，该方法用于在目录树上创建文件（创建目录使用mkdir），需要的参数比较多，包括文件名，权限，客户端名，是否覆盖已存在文件，副本数和块大小。NameNode的create调用NameSystem的startFile方法（startFile需要的参数clientMachine从线程局部变量获取）。
startFile方法先调用startFileInternal完成操作，然后调用logSync，等待日志写完后才返回。
startFileInternal不但服务于startFile，也被appendFile调用（通过参数append区分）。方法的最开始是一堆检查，包括：安全模式，文件名src是否正确，权限，租约，replication参数，overwrite参数（对append操作是判断src指向是否存在并且是文件）。租约检查很简单，如果通过FSDirectory.getFileINode(src)得到的文件是出于构造状态，表明有客户正在操作该文件，这时会抛出异常AlreadyBeingCreatedException。
如果对于创建操作，会通过FSDirectory的addFile往目录树上添加一个文件并在租约管理器里添加一条记录。
对于append操作，执行的是构造一个新的INodeFileUnderConstruction并替换原有的节点，然后在租约管理器里添加一条记录。
总的来说，最简单的create流程就是在目录树上创建一个INodeFileUnderConstruction对象并往租约管理器里添加一条记录。

我们顺便分析一下append吧，它的返回值是LocatedBlock，比起getBlockLocations，它只需要返回数组的一项。appendFile是NameSystem的实现方法，它首先调用上面讨论的startFileInternal方法（已经在租约管理器里添加了一条记录）然后写日志。然后寻找对应文件INodeFile中记录的最后一个block，并通过BlocksMap.getStoredBlock()方法得到BlockInfo，然后再从BlocksMap中获得所有的DatanodeDescriptor，就可以构造LocatedBlock了。需要注意的，如果该Block在需要被复制的集合（UnderReplicatedBlocks）中，移除它。
如果文件刚被创建或者是最后一个数据块已经写满，那么append会返回null，这是客户端需要使用addBlock，为文件添加数据块。


-------------------------------------------------------------------------------------------------------------------------------------------------

public boolean setReplication(String src,
                                      short replication
                                      ) throws IOException;

setReplication，设置文件src的副本数为replication，返回值为boolean，在FSNameSystem中，调用方法setReplicationInternal，然后写日志。
setReplicationInternal上来自然是检查参数了，然后通过FSDirectory的setReplication，设置新的副本数，并获取老的副本数。根据新旧数，决定删除/复制数据块。
增加副本数通过调用updateNeededReplications，为了获取UnderReplicatedBlocks. update需要的参数，FSNameSystem提供了内部方法countNodes和getReplication，获得对应的数值（这两个函数都很简单）。
proccessOverReplicatedBlock用于减少副本数，它被多个方法调用：



主要参数有block，副本数，目标DataNode，源DataNode（用于删除）。proccessOverReplicatedBlock首先找出block所在的，处于非Decommission状态的DataNode的信息，然后调用chooseExcessReplicates。chooseExcessReplicates执行：

l          按机架位置，对DatanodeDescriptor进行分组；

l          将DataNode分为两个集合，分别是一个机架包含一个以上的数据块的和剩余的；

l          选择可以删除的数据块（顺序是：源DataNode，同一个机架上的，剩余的），把它加到recentInvalidateSets中。

  public void setPermission(String src,FsPermission permission
                                 ) throwsIOException;

setPermission，用于设置文件的访问权限。非常简单，首先检查是否有权限，然后调用FSDirectory.setPermission修改文件访问权限。

  public void setOwner(String src, Stringusername, String groupname
      ) throws IOException;

  public voidsetTimes(String src, long mtime, long atime) throws IOException;

  public void setQuota(String path, long namespaceQuota, long diskspaceQuota)
                      throws IOException;

setOwner，设置文件的文件主和文件组，setTimes，设置文件的访问时间和修改时间，setQuota,设置某路径的空间限额和空间额度，和setPermission类似，调用FSDirectory的对应方法，简单。

  public booleansetSafeMode(FSConstants.SafeModeAction action) throws IOException;

前面我们已经介绍了NameNode的安全模式，客户端通过上面的方法，可以让NameNode进入（SAFEMODE_ENTER）/退出（SAFEMODE_LEAVE）安全模式或查询（SAFEMODE_GET）状态。FSNamesystem的setSafeMode处理这个命令，对于进入安全模式的请求，如果系统现在不处于安全模式，那么创建一个SafeModeInfo对象（创建的这个对象有别于启动时创建的那个SafeModeInfo，它不会自动退出，因为threshold=1.5f），这标志着系统进入安全模式。退出安全模式很简单，将safeMode赋空就可以啦。

  public FileStatus[]getListing(String src) throwsIOException;

分析完set*以后，我们来看get*。getListing对应于UNIX系统的ls命令，返回值是FileStatus数组，FileStatus的类图如下，它其实给出了文件的详细信息，如大小，文件主等等。其实，这些信息都存在INode*中，我们只需要把这些信息搬到FileStatus中就OK啦。FSNamesystem和FSDirectory中都有同名方法，真正干活的地方在FSDirectory中。getListing不需要写日志。

public long[] getStats() throws IOException;

getStatus得到的是文件系统的信息，UNIX对应命令为du，它的实现更简单，所有的信息都存放在FSNamesystem对象里。

  public DatanodeInfo[] getDatanodeReport(FSConstants.DatanodeReportTypetype)

  throws IOException;

getDatanodeReport，获取当前DataNode的状态，可能的选项有DatanodeReportType.ALL, IVE和DEAD。FSNamesystem的同名方法调用getDatanodeListForReport，通过HostsFileReader读取对应信息。

  public long getPreferredBlockSize(Stringfilename) throwsIOException;

getPreferredBlockSize，返回INodeFile.preferredBlockSize，数据块大小。

  public FileStatusgetFileInfo(String src) throwsIOException;

和getListing类似，不再分析。
  publicContentSummary getContentSummary(String path) throws IOException;
得到文件树的一些信息，如下图：

public void metaSave(String filename) throws IOException;

这个也很简单，它把系统的metadata输出/添加到指定文件上（NameNode所在的文件系统）。


---------------------------------------------------------------------------------------------------------------------------------------------------



软柿子都捏完了，我们开始啃硬骨头。前面已经分析过getBlockLocations，create，append，setReplication，setPermission和setOwner，接下来我们继续回来讨论和文件内容相关的操作。

  public void abandonBlock(Block b, Stringsrc, String holder
      ) throws IOException;

abandonBlock用于放弃一个数据块。普通的文件系统中并没有“放弃”操作，HDFS出现放弃数据块的原因，如下图所示。当客户端通过其他操作（如下面要介绍的addBlock方法）获取LocatedBlock后，可以打开到一个block的输出流，由于从DataNode出错到NameNode发现这个信息，需要有一段时间（NameNode长时间收到DataNode心跳），打开输出流可能出错，这时客户端可以向NameNode请求放弃这个数据块。






abandonBlock的处理不是很复杂，首先检查租约（调用checkLease方法。block对应的文件存在，文件处于构造状态，租约拥有者匹配），如果通过检查，调用FSDirectory的removeBlock，从INodeFileUnderConstruction/BlocksMap/CorruptReplicasMap中删除block，然后通过logOpenFile()记录变化（logOpenFile真是万能啊）。

  public LocatedBlock addBlock(Stringsrc, String clientName) throwsIOException;

写HDFS的文件时，如果数据块被写满，客户端可以通过addBlock创建新的数据块。具体的创建工作由FSNamesystem的getAdditionalBlock方法完成，当然上来就是一通检查（是否安全模式，命名/存储空间限额，租约，数据块副本数，保证DataNode已经上报数据块状态），然后通过ReplicationTargetChooser，选择复制的目标（如果目标数不够副本数，又是一个异常），然后，就可以分配数据块了。allocateBlock创建一个新的Block对象，然后调用addBlock，检查参数后把数据块加到BlocksMap对象和对应的INodeFile对象中。allocateBlock返回后，getAdditionalBlock还会继续更新一些需要记录的信息，最后返回一个新构造的LocatedBlock。

  public boolean complete(String src, StringclientName) throwsIOException;

当客户端完成对数据块的写操作后，调用complete完成写操作。方法complete如果返回是false，那么，客户端需要继续调用complete方法。
FSNamesystem的同名方法调用completeFileInternal，它会：

l          检查环境；

l          获取src对应的INode；

l          如果INode存在，并且处于构造状态，获取数据块；

l          如果获取数据块返回空，返回结果CompleteFileStatus.OPERATION_FAILED，FSNamesystem的complete会抛异常返回；

l          如果上报文件完成的DataNode数不够系统最小的副本数，返回STILL_WAITING；

l          调用finalizeINodeFileUnderConstruction；

l          返回成功COMPLETE_SUCCESS

其中，对finalizeINodeFileUnderConstruction的处理包括：

l          释放租约；

l          将对应的INodeFileUnderConstruction对象转换为INodeFile对象，并在FSDirectory进行替换；

l          调用FSDirectory.closeFile关闭文件，其中会写日志logCloseFile(path, file)。

l          检查副本数，如果副本数小于INodeFile中的目标数，那么添加数据块复制任务。

我们可以看到，complete一个文件还是比较复杂的，需要释放很多的资源。

  public void reportBadBlocks(LocatedBlock[]blocks) throwsIOException;

调用reportBadBlocks的地方比较多，客户端可能调用，DataNode上也可能调用。











由于上报的是个数组，reportBadBlocks会循环处理，调用FSNamesystem的markBlockAsCorrupt方法。markBlockAsCorrupt方法需要两个参数，blk（数据块）和dn（所在的DataNode信息）。如果系统目前副本数大于要求，那么直接调用invalidateBlock方法。方法invalidateBlock很简单，在检查完系统环境以后，先调用addToInvalidates方法往FSNamesystem.recentInvalidateSets添加一项，然后调用removeStoredBlock方法。removeStoredBlock被多个方法调用，它会执行：l          从BlocksMap中删除记录removeNode(block,node)；
l          如果目前系统中还有其他副本，调用decrementSafeBlockCount（可能的调整安全模式参数）和updateNeededReplications（跟新可能存在的block复制信息，例如，现在系统中需要复制1个数据块，那么更新后，需要复制2个数据块）；
l          如果目前系统中有多余数据块等待删除（在excessReplicateMap中），那么移除对应记录；
l          删除在CorruptReplicasMap中的记录（可能有）。




removeStoredBlock其实也是涉及了多处表操作，包括BlocksMap，excessReplicateMap和CorruptReplicasMap。我们回到markBlockAsCorrupt，如果系统目前副本数小于要求，那么很显然，我们需要对数据块进行复制。首先将现在的数据块加入到CorruptReplicasMap中，然后调用updateNeededReplications，跟新复制信息。markBlockAsCorrupt这个流程太复杂了，我们还是画个图吧：

下一篇

Hadoop二次开发必备，Hadoop源码分析（十二）

上一篇

Hadoop二次开发必备，Hadoop源码分析（十）