HDFS2.X源码分析之：NameNode写文件原理

原文出自云台博客：http://yuntai.1kapp.com/?p=950
HDFS被设计成写一次，读多次的应用场景，这应该跟它的MapReduce机制是紧密关联的，通过对线上的读写比例监控，大概读写比是10:1，也验证了它设计的目标。
GFS论文提到的写入文件简单流程：



HDFS详细流程：



写入文件的过程比读取较为复杂：
1、
使用HDFS提供的客户端开发库Client，向远程的Namenode发起RPC请求；
2、
Namenode会检查要创建的文件是否已经存在，创建者是否有权限进行操作，成功则会为文件创建一个记录，否则会让客户端抛出异常；
3、
当客户端开始写入文件的时候，会将文件切分成多个packets，并在内部以数据队列"data queue"的形式管理这些packets，并向Namenode申请新的blocks，获取用来存储replicas的合适的datanodes列表，列表的大小根据在Namenode中对replication的设置而定；
4、
开始以pipeline（管道）的形式将packet写入所有的replicas中。开发库把packet以流的方式写入第一个datanode，该datanode把该packet存储之后，再将其传递给在此pipeline中的下一个datanode，直到最后一个datanode，这种写数据的方式呈流水线的形式；
5、
最后一个datanode成功存储之后会返回一个ack packet，在pipeline里传递至客户端，在客户端的开发库内部维护着"ack queue"，成功收到datanode返回的ack packet后会从"ack queue"移除相应的packet；
6、
如果传输过程中，有某个datanode出现了故障，那么当前的pipeline会被关闭，出现故障的datanode会从当前的pipeline中移除，剩余的block会继续剩下的datanode中继续以pipeline的形式传输，同时Namenode会分配一个新的datanode，保持replicas设定的数量；

3.2.1.创建元数据

下图显示了NameNode在写入文件过程中创建文件元数据流程图，主要分为如下一个步骤：



创建元数据流程图
1、 获取RPC传递过来的参数，并进行相关合法性检查；
2、 验证父目录是否存在，如果不存在就创建（如果客户端要求如果父目录不存在就创建，否会报父目录不存在）；
3、 判断该操作是覆盖以前的旧文件，还是追加写，还是重新创建一个文件；如果覆盖以前旧文件，删除旧文件元数据及其DN等映射关系，如果追加写就修改以前inode类型为UC型inode，如果重新创建文件需要重新创建inode对象，当然该对象会挂到tree结构上去；同时也别忘了将其加入租约；
4、 将操作写入日志文件；
其代码调用流程为：NameNodeRpcServer. Create()->FSNamesystem. startFile()->FSNamesystem.startFileInt()->FSNamesystem . checkOperation ()->FSNamesystem .startFileInternal(),其中startFileInternal具体实现如下：

。。。。。。。。。。。。。。。。。。。。。。。。。。。。。。。。。。。
try {
//从tree查找该inode是否存在
INodeFile myFile = dir.getFileINode(src);
try {
blockManager.verifyReplication(src, replication, clientMachine);
} catch(IOException e) {
throw new IOException("failed to create "+e.getMessage());
}
boolean create = flag.contains(CreateFlag.CREATE);
if (myFile == null) {
if (!create) {
throw new FileNotFoundException("failed to overwrite or append to non-existent file "
+ src + " on client " + clientMachine);
}
} else {
// File exists - must be one of append or overwrite
//如果文件存在，且需要覆盖之前文件，先将之前文件删除
if (overwrite) {
delete(src, true);
} else {
//打开一个存在的文件写，可能需要恢复lease
// Opening an existing file for write - may need to recover lease.
recoverLeaseInternal(myFile, src, holder, clientMachine, false);

if (!append) {
throw new FileAlreadyExistsException("failed to create file " + src
+ " on client " + clientMachine
+ " because the file exists");
}
}
}
final DatanodeDescriptor clientNode =
blockManager.getDatanodeManager().getDatanodeByHost(clientMachine);
if (append && myFile != null) {
return prepareFileForWrite(
src, myFile, holder, clientMachine, clientNode, true);
} else {
// Now we can add the name to the filesystem. This file has no
// blocks associated with it.
//
checkFsObjectLimit();
// increment global generation stamp
long genstamp = nextGenerationStamp();
INodeFileUnderConstruction newNode = dir.addFile(src, permissions,
replication, blockSize, holder, clientMachine, clientNode, genstamp);
if (newNode == null) {
throw new IOException("DIR* NameSystem.startFile: " +
"Unable to add file to namespace.");
}
leaseManager.addLease(newNode.getClientName(), src);
// record file record in log, record new generation stamp
getEditLog().logOpenFile(src, newNode);
if (NameNode.stateChangeLog.isDebugEnabled()) {
NameNode.stateChangeLog.debug("DIR* NameSystem.startFile: "
+"add "+src+" to namespace for "+holder);
}

3.2.2.分配数据块

当Client接收到NameNode创建成功元数据返回结果后，会请求NameNode分配块，对于文件的每个块，Client均会RPC调用NameNode端的addbrick操作，在addbrick内，会为新块产生块ID与时间戳，同时还有该块的几个副本（默认3）存储在应该存储咋哪几台DataNode。对于NameNode新分配的块，会经历如下几个状态：



随便提一下，对于一个块的副本，会经过如下几个状态：



Client接收到分配块，上传数据块到响应节点的过程中，如果失败，则会重新请求NameNode分配块，当然下次分配的块再也不会分配到上次块所分配的DataNode节点上了，因为上次Upload数据块已经失败了，可能该DataNode节点与client之间的网络出现异常。



分配数据块流程图
代码调用流程：
NameNodeRpcServer.addBlock()->FSNamesystem.getAdditionalBlock(),而整个过程的核心实现也是在方法getAdditionalBlock内部，细节可以看下面的代码，这儿主要提一下分配块相对重要和容易让人费解的地方。在分配块之前，会比较文件的最后一个块信息与上次client上传的块是否一致，如果一致我们是可以理解的，可能有如下两种情况：
1、 再请求本次块分配前，NameNode刚给文件分配了块，所以本次比较应该是相等的；
2、 如果文件为新文件，则最后一个块与上次Client上传的块均为空，所以也是相等的，相当于是分配第一个块；
但是也也有可能上次client上传的块与获得的文件的最后一个数据块信息不相同：
1、 本次操作为刚打开一个文件append的第一个操作，因此lastBlockInFile不为空，但是previousBlock为空；
2、 可能在client进行本次请求前，由于客户端timeout或者HAfailover，没有接收到分配数据块信息成功的响应，因此客户端再次请求分配块，但是这样的话，对于文件来说，块就重复了。



请求timeout流程图



NameNode Failover情况流程图

目前Hadoop2.0的做法是如果文件的最后一个块大小为0，抛弃该种分配的数据块并且重新分配，但是此是在实现上可能不是特别好，首先抛弃该文件的最后一个块，并更新最新的数据块信息到日志，后面有更新了一次最新块信息到日志，两次存在重复更新的嫌疑，代码片段：

//如果最后一个块已经分配，但是大小为0，这删除最后一个块，可否不重新分配呢？？这种时候可能需要修改客户端实现了，
//因此为出现该问题情况只有在异常情况下才发生，所以多分配一次对整个分配块过程性能没什么影响
// The retry case ("b" above) -- abandon the old block.
NameNode.stateChangeLog.info("BLOCK* NameSystem.allocateBlock: " +
"caught retry for allocation of a new block in " +
src + ". Abandoning old block " + lastBlockInFile);
//抛弃该文件的最后一个块，并更新快信息到日志
dir.removeBlock(src, pendingFile, lastBlockInFile);
//在写一次path及其块列表到日志，有疑问的是，不是上面一部也写块信息到日志了么，相当于上面已经更新过护具数据块信息了
dir.persistBlocks(src, pendingFile);

3、 除去1,2两种情况，目前处理为抛出异常。

下面贴出为文件分配块通过机架选择DataNode节点，建立文件与新块之间的映射，和将更新后的块信息写入日志的代码片段

// choose targets for the new block to be allocated.
final DatanodeDescriptor targets[] = blockManager.chooseTarget(
src, replication, clientNode, excludedNodes, blockSize);

// Allocate a new block and record it in the INode.
writeLock();
try {
checkOperation(OperationCategory.WRITE);
if (isInSafeMode()) {
throw new SafeModeException("Cannot add block to " + src, safeMode);
}
INode[] pathINodes = dir.getExistingPathINodes(src);
int inodesLen = pathINodes.length;
checkLease(src, clientName, pathINodes[inodesLen-1]);
INodeFileUnderConstruction pendingFile  = (INodeFileUnderConstruction)
pathINodes[inodesLen - 1];

if (!checkFileProgress(pendingFile, false)) {
throw new NotReplicatedYetException("Not replicated yet:" + src);
}

// allocate new block record block locations in INode.
//为该文件分配一个新块
newBlock = allocateBlock(src, pathINodes, targets);

for (DatanodeDescriptor dn : targets) {
dn.incBlocksScheduled();
}
//写日志，更新块信息
dir.persistBlocks(src, pendingFile);
} finally {
writeUnlock();
}
//为什么此处还需要同步一下？因为在写日志操作的时候本身就已经需要同步，难道是如果persistBlocks为true，
//则强制快速同步一次,因为虽然写日志会调用同步，但是有条件的
if (persistBlocks) {
getEditLog().logSync();
}

这样就分配成功了数据块，并且将响应信息返回给客户端。