一、文件的打开
1.1、客户端
HDFS打开一个文件，需要在客户端调用DistributedFileSystem.open(Path f, int bufferSize)，其实现为：
|
public FSDataInputStream open(Path f, int bufferSize) throws IOException { return new DFSClient.DFSDataInputStream( dfs.open(getPathName(f), bufferSize, verifyChecksum, statistics)); } |
其中dfs为DistributedFileSystem的成员变量DFSClient，其open函数被调用，其中创建一个DFSInputStream(src, buffersize, verifyChecksum)并返回。
在DFSInputStream的构造函数中，openInfo函数被调用，其主要从namenode中得到要打开的文件所对应的blocks的信息，实现如下：
|
synchronized void openInfo() throws IOException { LocatedBlocks newInfo = callGetBlockLocations(namenode, src, 0, prefetchSize); this.locatedBlocks = newInfo; this.currentNode = null; } |
|
private static LocatedBlocks callGetBlockLocations(ClientProtocol namenode, String src, long start, long length) throws IOException { return namenode.getBlockLocations(src, start, length); } |
LocatedBlocks主要包含一个链表的List<LocatedBlock> blocks，其中每个LocatedBlock包含如下信息：
- Block b：此block的信息
- long offset：此block在文件中的偏移量
- DatanodeInfo[] locs：此block位于哪些DataNode上
上面namenode.getBlockLocations是一个RPC调用，最终调用NameNode类的getBlockLocations函数。
1.2、NameNode
NameNode.getBlockLocations实现如下：
|
public LocatedBlocks getBlockLocations(String src, long offset, long length) throws IOException { return namesystem.getBlockLocations(getClientMachine(), src, offset, length); } |
namesystem是NameNode一个成员变量，其类型为FSNamesystem，保存的是NameNode的name space树，其中一个重要的成员变量为FSDirectory dir。
FSDirectory 和Lucene中的FSDirectory没有任何关系，其主要包括FSImage fsImage，用于读写硬盘上的fsimage文件，FSImage类有成员变量FSEditLog editLog，用于读写硬盘上的edit文件，这两个文件的关系在上一篇文章中已经解释过。
FSDirectory还有一个重要的成员变量INodeDirectoryWithQuota rootDir，INodeDirectoryWithQuota的父类为INodeDirectory，实现如下：
|
public class INodeDirectory extends INode { …… private List<INode> children; …… ｝ |
由 此可见INodeDirectory本身是一个INode，其中包含一个链表的INode，此链表中，如果仍为文件夹，则是类型 INodeDirectory，如果是文件，则是类型INodeFile，INodeFile中有成员变量BlockInfo blocks[]，是此文件包含的block的信息。显然这是一棵树形的结构。
FSNamesystem.getBlockLocations函数如下：
|
public LocatedBlocks getBlockLocations(String src, long offset, long length, boolean doAccessTime) throws IOException { final LocatedBlocks ret = getBlockLocationsInternal(src, dir.getFileINode(src), offset, length, Integer.MAX_VALUE, doAccessTime); return ret; } |
dir.getFileINode(src)通过路径名从文件系统树中找到INodeFile，其中保存的是要打开的文件的INode的信息。
getBlockLocationsInternal的实现如下：
|
private synchronized LocatedBlocks getBlockLocationsInternal(String src, INodeFile inode, long offset, long length, int nrBlocksToReturn, boolean doAccessTime) throws IOException { //得到此文件的block信息 Block[] blocks = inode.getBlocks(); List<LocatedBlock> results = new ArrayList<LocatedBlock>(blocks.length); //计算从offset开始，长度为length所涉及的blocks int curBlk = 0; long curPos = 0, blkSize = 0; int nrBlocks = (blocks[0].getNumBytes() == 0) ? 0 : blocks.length; for (curBlk = 0; curBlk < nrBlocks; curBlk++) { blkSize = blocks[curBlk].getNumBytes(); if (curPos + blkSize > offset) { //当offset在curPos和curPos + blkSize之间的时候，curBlk指向offset所在的block break; } curPos += blkSize; } long endOff = offset + length; //循环，依次遍历从curBlk开始的每个block，直到当前位置curPos越过endOff do { int numNodes = blocksMap.numNodes(blocks[curBlk]); int numCorruptNodes = countNodes(blocks[curBlk]).corruptReplicas(); int numCorruptReplicas = corruptReplicas.numCorruptReplicas(blocks[curBlk]); boolean blockCorrupt = (numCorruptNodes == numNodes); int numMachineSet = blockCorrupt ? numNodes : (numNodes - numCorruptNodes); //依次找到此block所对应的datanode，将其中没有损坏的放入machineSet中 DatanodeDescriptor[] machineSet = new DatanodeDescriptor[numMachineSet]; if (numMachineSet > 0) { numNodes = 0; for(Iterator<DatanodeDescriptor> it = blocksMap.nodeIterator(blocks[curBlk]); it.hasNext();) { DatanodeDescriptor dn = it.next(); boolean replicaCorrupt = corruptReplicas.isReplicaCorrupt(blocks[curBlk], dn); if (blockCorrupt || (!blockCorrupt && !replicaCorrupt)) machineSet[numNodes++] = dn; } } //使用此machineSet和当前的block构造一个LocatedBlock results.add(new LocatedBlock(blocks[curBlk], machineSet, curPos, blockCorrupt)); curPos += blocks[curBlk].getNumBytes(); curBlk++; } while (curPos < endOff && curBlk < blocks.length && results.size() < nrBlocksToReturn); //使用此LocatedBlock链表构造一个LocatedBlocks对象返回 return inode.createLocatedBlocks(results); } |
1.3、客户端
通过RPC调用，在NameNode得到的LocatedBlocks对象，作为成员变量构造DFSInputStream对象，最后包装为FSDataInputStream返回给用户。
二、文件的读取
2.1、客户端
文件读取的时候，客户端利用文件打开的时候得到的FSDataInputStream.read(long position, byte[] buffer, int offset, int length)函数进行文件读操作。
FSDataInputStream会调用其封装的DFSInputStream的read(long position, byte[] buffer, int offset, int length)函数，实现如下：
|
public int read(long position, byte[] buffer, int offset, int length) throws IOException { long filelen = getFileLength(); int realLen = length; if ((position + length) > filelen) { realLen = (int)(filelen - position); } //首先得到包含从offset到offset + length内容的block列表 //比如对于64M一个block的文件系统来说，欲读取从100M开始，长度为128M的数据，则block列表包括第2，3，4块block List<LocatedBlock> blockRange = getBlockRange(position, realLen); int remaining = realLen; //对每一个block，从中读取内容 //对于上面的例子，对于第2块block，读取从36M开始，读取长度28M，对于第3块，读取整一块64M，对于第4块，读取从0开始，长度为36M，共128M数据 for (LocatedBlock blk : blockRange) { long targetStart = position - blk.getStartOffset(); long bytesToRead = Math.min(remaining, blk.getBlockSize() - targetStart); fetchBlockByteRange(blk, targetStart, targetStart + bytesToRead - 1, buffer, offset); remaining -= bytesToRead; position += bytesToRead; offset += bytesToRead; } assert remaining == 0 : "Wrong number of bytes read."; if (stats != null) { stats.incrementBytesRead(realLen); } return realLen; } |
其中getBlockRange函数如下：
|
private synchronized List<LocatedBlock> getBlockRange(long offset, long length) throws IOException { List<LocatedBlock> blockRange = new ArrayList<LocatedBlock>(); //首先从缓存的locatedBlocks中查找offset所在的block在缓存链表中的位置 int blockIdx = locatedBlocks.findBlock(offset); if (blockIdx < 0) { // block is not cached blockIdx = LocatedBlocks.getInsertIndex(blockIdx); } long remaining = length; long curOff = offset; while(remaining > 0) { LocatedBlock blk = null; //按照blockIdx的位置找到block if(blockIdx < locatedBlocks.locatedBlockCount()) blk = locatedBlocks.get(blockIdx); //如果block为空，则缓存中没有此block，则直接从NameNode中查找这些block，并加入缓存 if (blk == null || curOff < blk.getStartOffset()) { LocatedBlocks newBlocks; newBlocks = callGetBlockLocations(namenode, src, curOff, remaining); locatedBlocks.insertRange(blockIdx, newBlocks.getLocatedBlocks()); continue; } //如果block找到，则放入结果集 blockRange.add(blk); long bytesRead = blk.getStartOffset() + blk.getBlockSize() - curOff; remaining -= bytesRead; curOff += bytesRead; //取下一个block blockIdx++; } return blockRange; } |
其中fetchBlockByteRange实现如下：
|
private void fetchBlockByteRange(LocatedBlock block, long start, long end, byte[] buf, int offset) throws IOException { Socket dn = null; int numAttempts = block.getLocations().length; //此while循环为读取失败后的重试次数 while (dn == null && numAttempts-- > 0 ) { //选择一个DataNode来读取数据 DNAddrPair retval = chooseDataNode(block); DatanodeInfo chosenNode = retval.info; InetSocketAddress targetAddr = retval.addr; BlockReader reader = null; try { //创建Socket连接到DataNode dn = socketFactory.createSocket(); dn.connect(targetAddr, socketTimeout); dn.setSoTimeout(socketTimeout); int len = (int) (end - start + 1); //利用建立的Socket链接，生成一个reader负责从DataNode读取数据 reader = BlockReader.newBlockReader(dn, src, block.getBlock().getBlockId(), block.getBlock().getGenerationStamp(), start, len, buffersize, verifyChecksum, clientName); //读取数据 int nread = reader.readAll(buf, offset, len); return; } finally { IOUtils.closeStream(reader); IOUtils.closeSocket(dn); dn = null; } //如果读取失败，则将此DataNode标记为失败节点 addToDeadNodes(chosenNode); } } |
BlockReader.newBlockReader函数实现如下：
|
public static BlockReader newBlockReader( Socket sock, String file, long blockId, long genStamp, long startOffset, long len, int bufferSize, boolean verifyChecksum, String clientName) throws IOException { //使用Socket建立写入流，向DataNode发送读指令 DataOutputStream out = new DataOutputStream( new BufferedOutputStream(NetUtils.getOutputStream(sock,HdfsConstants.WRITE_TIMEOUT))); out.writeShort( DataTransferProtocol.DATA_TRANSFER_VERSION ); out.write( DataTransferProtocol.OP_READ_BLOCK ); out.writeLong( blockId ); out.writeLong( genStamp ); out.writeLong( startOffset ); out.writeLong( len ); Text.writeString(out, clientName); out.flush(); //使用Socket建立读入流，用于从DataNode读取数据 DataInputStream in = new DataInputStream( new BufferedInputStream(NetUtils.getInputStream(sock), bufferSize)); DataChecksum checksum = DataChecksum.newDataChecksum( in ); long firstChunkOffset = in.readLong(); //生成一个reader，主要包含读入流，用于读取数据 return new BlockReader( file, blockId, in, checksum, verifyChecksum, startOffset, firstChunkOffset, sock ); } |
BlockReader的readAll函数就是用上面生成的DataInputStream读取数据。
2.2、DataNode
在DataNode启动的时候，会调用函数startDataNode，其中与数据读取有关的逻辑如下：
|
void startDataNode(Configuration conf, AbstractList<File> dataDirs ) throws IOException { …… // 建立一个ServerSocket，并生成一个DataXceiverServer来监控客户端的链接 ServerSocket ss = (socketWriteTimeout > 0) ? ServerSocketChannel.open().socket() : new ServerSocket(); Server.bind(ss, socAddr, 0); ss.setReceiveBufferSize(DEFAULT_DATA_SOCKET_SIZE); // adjust machine name with the actual port tmpPort = ss.getLocalPort(); selfAddr = new InetSocketAddress(ss.getInetAddress().getHostAddress(), tmpPort); this.dnRegistration.setName(machineName + ":" + tmpPort); this.threadGroup = new ThreadGroup("dataXceiverServer"); this.dataXceiverServer = new Daemon(threadGroup, new DataXceiverServer(ss, conf, this)); this.threadGroup.setDaemon(true); // auto destroy when empty …… } |
DataXceiverServer.run()函数如下：
|
public void run() { while (datanode.shouldRun) { //接受客户端的链接 Socket s = ss.accept(); s.setTcpNoDelay(true); //生成一个线程DataXceiver来对建立的链接提供服务 new Daemon(datanode.threadGroup, new DataXceiver(s, datanode, this)).start(); } try { ss.close(); } catch (IOException ie) { LOG.warn(datanode.dnRegistration + ":DataXceiveServer: " + StringUtils.stringifyException(ie)); } } |
DataXceiver.run()函数如下：
|
public void run() { DataInputStream in=null; try { //建立一个输入流，读取客户端发送的指令 in = new DataInputStream( new BufferedInputStream(NetUtils.getInputStream(s), SMALL_BUFFER_SIZE)); short version = in.readShort(); boolean local = s.getInetAddress().equals(s.getLocalAddress()); byte op = in.readByte(); // Make sure the xciver count is not exceeded int curXceiverCount = datanode.getXceiverCount(); long startTime = DataNode.now(); switch ( op ) { //读取 case DataTransferProtocol.OP_READ_BLOCK: //真正的读取数据 readBlock( in ); datanode.myMetrics.readBlockOp.inc(DataNode.now() - startTime); if (local) datanode.myMetrics.readsFromLocalClient.inc(); else datanode.myMetrics.readsFromRemoteClient.inc(); break; //写入 case DataTransferProtocol.OP_WRITE_BLOCK: //真正的写入数据 writeBlock( in ); datanode.myMetrics.writeBlockOp.inc(DataNode.now() - startTime); if (local) datanode.myMetrics.writesFromLocalClient.inc(); else datanode.myMetrics.writesFromRemoteClient.inc(); break; //其他的指令 …… } } catch (Throwable t) { LOG.error(datanode.dnRegistration + ":DataXceiver",t); } finally { IOUtils.closeStream(in); IOUtils.closeSocket(s); dataXceiverServer.childSockets.remove(s); } } |
|
private void readBlock(DataInputStream in) throws IOException { //读取指令 long blockId = in.readLong(); Block block = new Block( blockId, 0 , in.readLong()); long startOffset = in.readLong(); long length = in.readLong(); String clientName = Text.readString(in); //创建一个写入流，用于向客户端写数据 OutputStream baseStream = NetUtils.getOutputStream(s, datanode.socketWriteTimeout); DataOutputStream out = new DataOutputStream( new BufferedOutputStream(baseStream, SMALL_BUFFER_SIZE)); //生成BlockSender用于读取本地的block的数据，并发送给客户端 //BlockSender有一个成员变量InputStream blockIn用于读取本地block的数据 BlockSender blockSender = new BlockSender(block, startOffset, length, true, true, false, datanode, clientTraceFmt); out.writeShort(DataTransferProtocol.OP_STATUS_SUCCESS); // send op status //向客户端写入数据 long read = blockSender.sendBlock(out, baseStream, null); …… } finally { IOUtils.closeStream(out); IOUtils.closeStream(blockSender); } } |
三、文件的写入
下面解析向hdfs上传一个文件的过程。
3.1、客户端
上传一个文件到hdfs，一般会调用DistributedFileSystem.create，其实现如下：
|
public FSDataOutputStream create(Path f, FsPermission permission, boolean overwrite, int bufferSize, short replication, long blockSize, Progressable progress) throws IOException { return new FSDataOutputStream (dfs.create(getPathName(f), permission, overwrite, replication, blockSize, progress, bufferSize), statistics); } |
其最终生成一个FSDataOutputStream用于向新生成的文件中写入数据。其成员变量dfs的类型为DFSClient，DFSClient的create函数如下：
|
public OutputStream create(String src, FsPermission permission, boolean overwrite, short replication, long blockSize, Progressable progress, int buffersize ) throws IOException { checkOpen(); if (permission == null) { permission = FsPermission.getDefault(); } FsPermission masked = permission.applyUMask(FsPermission.getUMask(conf)); OutputStream result = new DFSOutputStream(src, masked, overwrite, replication, blockSize, progress, buffersize, conf.getInt("io.bytes.per.checksum", 512)); leasechecker.put(src, result); return result; } |
其中构造了一个DFSOutputStream，在其构造函数中，同过RPC调用NameNode的create来创建一个文件。
当然，构造函数中还做了一件重要的事情，就是streamer.start()，也即启动了一个pipeline，用于写数据，在写入数据的过程中，我们会仔细分析。
|
DFSOutputStream(String src, FsPermission masked, boolean overwrite, short replication, long blockSize, Progressable progress, int buffersize, int bytesPerChecksum) throws IOException { this(src, blockSize, progress, bytesPerChecksum); computePacketChunkSize(writePacketSize, bytesPerChecksum); try { namenode.create( src, masked, clientName, overwrite, replication, blockSize); } catch(RemoteException re) { throw re.unwrapRemoteException(AccessControlException.class, QuotaExceededException.class); } streamer.start(); } |
3.2、NameNode
NameNode的create函数调用namesystem.startFile函数，其又调用startFileInternal函数，实现如下：
|
private synchronized void startFileInternal(String src, PermissionStatus permissions, String holder, String clientMachine, boolean overwrite, boolean append, short replication, long blockSize ) throws IOException { ...... //创建一个新的文件，状态为under construction，没有任何data block与之对应 long genstamp = nextGenerationStamp(); INodeFileUnderConstruction newNode = dir.addFile(src, permissions, replication, blockSize, holder, clientMachine, clientNode, genstamp); ...... } |
3.3、客户端
下面轮到客户端向新创建的文件中写入数据了，一般会使用FSDataOutputStrea