搜索
HDFS读文件解析下图描述了在文件读过程中，client、NameNode和DataNode三者之间是如何互动的。1.client调用get方法得到HDFS文件系统的一个实例（DistributedFileSystem）。然后调用它的open方法。2.DistributedFileSystem通过RPC远程调用NameNode决定文件文件的block的位置信息。对于每一个bolck，NameNode返回block所在的DataNode（包括副本）的地址。DistributedFileSystem返回FSDataInputStream给client用来读数据。FSDataInputStream封装了DFSInputStream用于管理NameNode和DataNode的IO。3.client调用FSDataInputStream的read方法。4.DFSInputStream保存了block块所在的DataNode的地址信息。DFSInputStream连接第一个block的DataNode，readblock数据，传回给client。5.当第一个block读完，DFSInputStream关掉与这个DataNode的连接。然后开始第二个block。6.当client读结束，调用FSDataInputStream的close方法。在读的过程中，如果client和一个datanode通讯时出错，他会连接副本所在的datanode。这种client直接连接datanode读取数据的设计方法使HDFS可以同时相应很多client的同时并发。因为数据流量均匀的分布在所有的datanode上，NameNode只负责block的位置信息请求。HDFS写文件解析我们看一下创建文件，写文件最后关闭文件的流程。如下图：1.client通过调用DistributedFileSystem的create方法来创建文件。2.DistributedFileSystem通过RPC调用NameNode在文件系统的名字空间里创建一个文件，这个时候还没有任何block信息。DistributedFileSystem返回FSDataOutputStream给client。FSDataOutputStream封装了一个DFSOutputStream来处理与datanodes和namenode之间的通讯。3.当client写一个block数据的时候，DFSOutputStream把数据分成很多packet。FSDataOutputStream询问namenode挑选存储这个block以及它的副本的datanode列表。这个datanode列表组成了一个管道，在上图管道由三个datanode组成（备份参数是3），这三个datanode的选择有一定的副本放置策略，详细请看下一篇。4.FSDataOutputStream把packet写进管道的第一个datanode，然后管道把packet转发给第二个datanode，这样一直转发到最后一个datanode。5.只有当管道里所有datanode都返回写入成功，这个packet才算写成功，发送应答给FSDataOutputStream。开始下一个packet。如果某个datanode写失败了，会产生如下步骤，但是这些对client是透明的。1)管道关闭。2)正常的datanode上正在写的block会有一个新ID（需要和namenode通信）。这样失败的datanode上的那个不完整的block在上报心跳的时候会被删掉。3)失败的datanode会被移出管道。block中剩余的packet继续写入管道的其他两个datanode4)namenode会标记这个block的副本个数少于指定值。block的副本会稍后在另一个datanode创建。5)有些时候多个datanode会失败。只要dfs.replication.min(缺省是1)个datanode成功了，整个写入过程就算成功。缺少的副本会异步的恢复。6.当client完成了写所有block的数据后，调用FSDataOutputStream的close方法关闭文件。7.FSDataOutputStream通知namenode写文件结束。HDFS文件创建流程文件夹的创建是一个相对简单的过程，主要是通过FileSystem中的mkdirs（）方法，这个方法在DFSClient实例中调用同名方法mkdirs（），通过Hadoop本身的RPC机制调用Namenode的mkdirs（）方法，最终这个调用PUSH到FSNameSystem的mkdirsInternal方法，这个方法主要就是检验访问权限，最后通过FSDirectory的unprotectedMkdir（）方法，构建一个INodeDirectory实例添加到文件系统的目录树中。文件节点的创建与添加相对比较麻烦，主要步骤如下：FileSystem的create方法返回一个很重要的类FSDataOutputStream，这一点也比较好理解，就像java中的文件流一样，创建一个文件写入流对文件内容进行追加，首先我们看文件创建阶段namenode主要做了什么事情（权限验证以及租约验证这些前面都已经有提到，下面的内容就会掠过这一部分）DfSOutputStream在实例构建时，通过Hadoop本身的RPC机制调用Namenode的create方法，最终这个调用PUSH到FSNameSystem的StartFileInternal方法，需要做权限验证，租约检验等工作，这个方法主要作用就是创建一个INodeFileUnderConstruction实例（上面已经提过，文件写入过程中都会有一个INodeFileUnderConstruction与这个文件对应），这个实例最后通过FSDirectoty的addNode()方法添加到文件系统目录数中，这个时候文件创建操作就算完成了重要的第一步，文件系统中已经有了这个文件的记录。下面就涉及到文件的写入操作（相当复杂的部分）这个时候就需要用到返回的DfSOutputStream对象。这部分太复杂了，我们先分析一些基本模块儿，逐步吃透这部分的实现。整个分布式文件系统中网络通讯部分分为两类：1.命令类调用（这部分通过HADOOP的RPC机制进行支持）2.流式数据传输（这部分通过HADOOP的流式数据传输协议支持）为了保证数据的正确性，hadoop在多个关键处理单元做了数据检验操作，在流式数据网络传输部分通过校验和保证数据传输正常。Client在DfSOutputStream对象调用write方法时，系统并不会马上把数据写入SOCKET中，而是逐个构建Package并将这些Package加入一个队列。在DfSOutputStream对象构建时，系统通过Hadoop本身的RPC机制调用Namenode的create方法后，会启动一个后台线程streamer.start();这个线程的主要目的就是将上述的package队列写入SOCKET中。右图为DfSOutputStream的继承关系其中FSOutputSummer这个类其实质是一个decorator设计模式的实现，主要的目的就是在OutputStream的voidwrite(byteb[],intoff,intlen)方法中增加一些功能，上文已经提过，文件数据传输的同时，系统会在传输的数据中增加检验和数据，系统收到数据后对数据进行校验，保证数据传输的正确性,但是用户在对文件输出流进行操作的时候并不需要关注校验和数据，用户只需要不断的调用write方法在目标文件中追加数据。注：我们通常可以使用继承来实现功能的拓展,如果这些需要拓展的功能的种类很繁多,那么势必生成很多子类,增加系统的复杂性,同时,使用继承实现功能拓展,我们必须可预见这些拓展功能,这些功能是编译时就确定了,是静态的.使用Decorator的理由是:这些功能需要由用户动态决定加入的方式和时机.Decorator提供了即插即用的方法,在运行期间决定何时增加何种功能.我们看一下FSOutputSummer中write(byteb[],intoff,intlen)的实现publicsynchronizedvoidwrite(byteb[],intoff,intlen)throwsIOException{if(off0||len0||offb.length-len){thrownewArrayIndexOutOfBoundsException();}for(intn=0;nlen;n+=write1(b,off+n,len-n)){}}可以看到不断调用write1方法，保证数据发送的完整性。那么write1方法又做了什么事情呢，write1将用户需要写入的数据流首先写到自己的BUFFER中，达到一定数量（基本是一个chunk的大小）后进行CheckSum方法调用得到一段数据的校验和，然后通过writeChecksumChunk这个方法将数据以及该部分数据的校验和，按照一定格式要求一并写入Stream。writeChecksumChunk这个方法的主要作用就是将用户写入的数据以及该部分数据的校验和做为参数调用writeChunk()方法，这个方法是一个虚方法，真正的实现在DFSOutputStream这个类中，这也合情合理，本身FSOutputSummer这个类的作用仅仅是在输出流中增加校验和数据，至于数据是如何进行传输的是通过DFSOutputStream来实现的。那么接下来需要说明的就是DFSOutputStream的writeChunk这个方法了。HDFS流式数据网络传输的基本单位有哪些呢？chunk-package-block我们上文已经提过：等用户写入的数据达到一定数量（基本是一个chunk的大小）后就会对这段数据取校验和。一定数量的chunk就会组成一个package,这个package就是最终进行网络传输的基本单元，datanode收到package后，将这些package组合起来最终得到一个block。我们接下来通过实际主要的代码了解这部分功能的实现：currentPacket这个对象初始化的时候就是null,第一次写入数据时这个判断成立if(currentPacket==null){currentPacket=newPacket(packetSize,chunksPerPacket,bytesCurBlock);...//下面开始构建package包。//在package包中增加一个chunk，首先添加这个chunk所包含数据的checksumcurrentPacket.writeChecksum(checksum,0,cklen);currentPacket.writeData(b,offset,len);//然后添加这个chunk所包含的数据currentPacket.numChunks++;//增加这个package所包含的chunk个数bytesCurBlock+=len;//当前已经写入的byte个数//Ifpacketisfull,enqueueitfortransmission//如果这个package已经达到一定的chunk数量,准备实际的传输操作if(currentPacket.numChunks==currentPacket.maxChunks||bytesCurBlock==blockSize){......if(bytesCurBlock==blockSize)//如果用户写入的数据，已经达到一个block缺省大小（64M）{//设置当前的package是某一个block的最后一个packagecurrentPacket.lastPacketInBlock=true;//清除一些变量的值bytesCurBlock=0;lastFlushOffset=-1;}//这三段代码是关键的一部分代码，将已经构建完成的package写入一个dataQ