Hadoop读写过程详细代码

1、Hadoop学习摘要2:分析HDFS读写过程首先，打开文件1.1，客户端HDFS必须打开文件，然后在客户端上调用distributed file system . open(path f，int buffer size)，如下所示：public fsdata input stream open(path f，int buffer size)throws io exception return new DFS client . dfsdatasinputstream(Dfs.open (get pathname (f)、buffersize、verify checksum、statistics)

2、；其中DFS是DistributedFileSystem的成员变量，用于调用生成和返回DFS inputstream (src，buffersize，verify checksum)的open函数。在DFSInputStream的构造函数中，openInfo函数主要获取有关要在NameNode中打开的文件中的blocks的信息，如下所示：synchronized void open info()throws io exception localted blocks new info=callget block locations(name node，src，0，prefetch size)；t

3、his . located blocks=new info；This.currentNode=nullprivate static localted blocks callgetblocklocations(客户端协议名称节点、Stringsrc、long start、long length) throws ioexceptionreturn name node . getblocklocations(src、start、length)；LocatedBlocks主要包含一个链接的List blocks，每个LocatedBlock包含以下信息：Block b:有关此Block的信息长偏移：文

4、件中此块的偏移DatanodeInfo locs:此块位于哪个DataNode上上述NameNode.getBlockLocations是RPC调用，最终调用NameNode类的getBlockLocations函数。1.2，NameNodeNameNode.getBlockLocations的实现如下：public localted blocks get block locations(string src、长偏移、长长度)throws io exceptionreturn name system . getblocklocations(get client machine()、Src、of

5、fset、length)；Namesystem是类型为FSNamesystem的NameNode成员变量，用于存储NameNode的name space树，其中一个重要的成员变量是FSDirectory dir。FSDirectory和Lucene的FSDirectory没有关系。主要用于读取和写入硬盘上的FSImage文件(包括FSImage fsImage)，fsImage类具有成员变量fseditlog，用于读取和写入硬盘上的edit文件。这两个文件的关系在上一篇文章中讨论过。FSDirectory还具有一个名为INodeDirectoryWithQuota rootDir的重要成员变量

6、，INodeDirectory是实现的inodelivery的超类，如下所示：public class inode directory extends inode .Private List children.这是有关INodeDirectory本身是具有关联列表的索引节点，并且在此关联列表中仍然是文件夹的信息，是有关类型INode directory的信息，是文件的类型INodeFile，是INodeFile中包含成员变量BlockInfo blocks的文件中包含的块的信息。很明显，这是树形结构。FSNamesystem.getBlockLocations函数包括：public local

7、ted blocks get block locations(string src、long offset、long length、boolean do access time)throws io exception final localted blocks ret=getblocklocationsinternal(src，dir.get fileinode (src)、Offset、length、integer.max _ value、do access time)；Return retDir.getFileINode(src)通过路径名在文件系统树中查找索引节点文件。其中，存储有关要打

8、开的文件的索引节点的信息。GetBlockLocationsInternal的实现如下：专用同步锁定块getblocklocationsinternal(string src、INodeFile索引节点、长偏移、很长的路，Int nrBlocksToReturn、Boolean doAccessTime)Throws io exception/获取此文件的block信息blockblocks=inode . get blocks()；list results=new ArrayList(blocks . length)；/计算从offset开始，长度为与length相关的blocksint c

9、urBlk=0；Long curPos=0，blk size=0；Intnr blocks=(blocks 0)。getnumbytes()=0)？0 : blocks.lengthfor(curBlk=0；CurBlk nrBlocksCurBlk) Blk size=blocks curblk。getnumbytes()；曲面块大小偏移(if)/如果curPos和curPos blkSize之间存在offset，则curBlk指向offset所在的blockBreakCurPos=blkSizeLong endOff=offset length/循环显示从curBlk开始的每个block，

10、直到当前位置curPos超过endOff东北intnum nodes=block smap . num nodes(blockscurblk)；intnumcorruptnodes=count nodes(blockscurblk)。corrupt replicas()；intnumcorruptreplics=corrupt replicas . numcorruptreplics(blockscurblk)；boolean block corrupt=(numcorruptnodes=num nodes)；Int numMachineSet=blockCorrupt？NumNodes :(

11、num nodes-numCorruptNodes)；/找到与此block对应的datanode，然后将损坏插入到machineSetdatanodedescriptormachineset=new datanodedescriptornummachineset；If (numMachineSet 0) num nodes=0；For(Iterator it=block smap . node iterator(blockscurblk)；it . has next()；)DatanodeDescriptor dn=it . next()；boolean replica corrupt=cor

12、rupt replicas . isreplicatiuppt(blockscurblk，dn)；If (blockCorrupt |(！BlockCorrupt！ReplicaCorrupt)machine setnum nodes=dn；/使用此machineSet和当前block配置LocatedBlockresults . add(new localted block(blockscurblk、machineset、curpos、BlockCorrupt)Curpos=blocks curblk。getnumbytes()；CurBlk while (curPos endOffCurB

13、lk blocks.lengthresults . size()nrBlocksToReturn)；/使用此LocatedBlock links配置LocatedBlocks对象的返回Return inode。createlocated blocks(results)；1.3，客户端通过RPC调用从NameNode获取的LocatedBlocks对象将DFSInputStream对象配置为成员变量，然后包装到FSDataInputStream中并返回给用户。二、读取文件2.1，客户端读取文件时，客户端将打开文件时获取的fs data inputstream . read(long positi

14、on，byte buffer，int offset，int length)函数用于读取文件操作。FSDataInputStream通过调用封装的DFSInputStream的read (long position，byte buffer，int offset，int length)函数实现，如下所示：Public intread (long position，byte buffer，int offset，int length)Throws io exceptionlong filelen=getFileLength()；Int realLen=length位置长度文件(if)RealLen=(

15、int)(filelen-position)；/首先获取包含offset到offset length内容的block列表/例如，对于64米的单个块文件系统，要读取100米到128米长度的数据，块列表包含块2、3和4list block range=getblock range(position，re Allen)；Int remaining=realLen/读取每个block的内容/在上例中，块2的读取长度为36米，读取长度为28米，块3的读取长度为64米，块4的读取长度为0到36米，总计为128米锁定块3360块范围(for)long target start=position-blk .

16、getstartoffset()；long bytes to read=math . min(remaining，blk . get blocksize()-target start)；Fetchblockbytesrange (blk、targetstart、Targetstart bytes to read-1，buffer，offset)；remaining-=bytesToRead；职位=bytesToReadOffset=bytesToReadassert remaining=0 : wrong number of bytes read . ；If(状态！=null) stats . increment bytes read(re Allen)；Return realLen其中ge