Hadoop源码分析(5HDFS数据流).docx

Hadoop源码分析（5 HDFS数据流）作者：张孟志 日期：2013-07-30下面我们可以来开始分析DataNode上的动态行为。首先我们来分析DataXceiverServer和DataXceiver。DataNode上数据块的接受/发送并没有采用我们前面介绍的RPC机制，原因很简单，RPC是一个命令式的接口，而DataNode处理数据部分，往往是一种流式机制。DataXceiverServer和DataXceiver就是这个机制的实现。其中，DataXceiver还依赖于两个辅助类：BlockSender和BlockReceiver。下面是类图：DataXceiverServer很简单，它打开一个端口，然后每接收到一个连接，就创建一个DataXceiver，服务于该连接，并记录该连接的socket，对应的实现在DataXceiverServer的run方法里。当系统关闭时，DataXceiverServer将关闭监听的socket和所有DataXceiver的socket，这样就导致了DataXceiver出错并结束线程。DataXceiver才是真正干活的地方，目前，DataXceiver支持的操作总共有六条，分别是：OP_WRITE_BLOCK (80)：写数据块OP_READ_BLOCK (81)：读数据块OP_READ_METADATA (82)：读数据块元文件OP_REPLACE_BLOCK (83)：替换一个数据块OP_COPY_BLOCK (84)：拷贝一个数据块OP_BLOCK_CHECKSUM (85)：读数据块检验码DataXceiver首先读取客户端的版本号并检验，然后再读取一个字节的操作码，并转入相关的子程序进行处理。下面以$HADOOP_HOME/bin/hadoop fs -put 或$HADOOP_HOME/bin/hadoop fs -copyFromLocal （这两个命令是一样的）命令来介绍整个写入的流程(OP_WRITE_BLOCK (80)：写数据块)。客户端首先向namenode申请写入文件（这里仅讨论新写入一个文件的情况，目前Hadoop也支持append写入方式，但暂不讨论）；namenode接收请求后仅在namdnode端创建一个无对应block的文件并在整个hdfs中维护；接着datanode建立以本地要上传的文件作为输入流，namenode返回的路径作为输出流，利用IOUtils.copyBytes()函数开始上传数据文件；在client准备好发送、响应的两个队列和数据的packet后，向namenode申请对应的datenodes和blocks；namenode为其分配datanodes和创建新的blocks；然后由client建立到第一个datanode的连接（默认会有3份备份）开始正式写入数据，并由每个datanode建立到下一个datanode的连接完成数据备份直到最后一个datanode；同时每个datanode接受下一个datanode写入数据是否成功的响应（ACK），并最终传给client（以上过程类似于管线），释放资源或对错误进行处理。数据流图如图所示：步骤一：客户端通过对DistributedFileSystem对象调用create()方法来创建文件； 步骤二：DistributedFileSystem对namenode创建一个RPC调用，在文件系统的命名空间中创建一个新文件。此时该文件中没有对应的数据块；namenode执行各种检查以确保这个文件不存在，并且客户端有创建该文件的权限； 步骤三：DistributedFileSystem向客户端返回一个负责处理datanode和namenode间通信的FSDataOutputStream对象，由此客户端开始写入数据。DFSOutputStream将文件分成一个个数据包，并写入称为数据队列(data queue)的内部队列； 步骤四： DataStreamer处理数据队列，根据datenode列表来要求namenode分配适合的新块来存储数据备份。一组datanode组成一条管线(pipeline)。DataStreamer将数据包流式传输到管线中的第1个datanode，再根据管线节点顺序传输数据包。 步骤五：DFSOutputStream维护一个称为确认队列(ack queue)的内部数据包队列等待datanode的收到确认回执。收到管道中所有datanode确认信息后，该数据包从确认队列删除。 步骤六、步骤七：客户端完成数据的写入后，会对数据流调用close()方法。该操作将剩余的所有数据包写入datanode管线中，并在联系namenode且发送文件写入完成信号之前，等待确认。Namenode以知道文件由哪些块组成（步骤四），所以在返回成功前只需要等待数据块进行最小量的复制。下面就来详细介绍以上过程在hadoop中的代码实现：一、client端申请org.apache.hadoop.fs. FsShell:public int run(String argv) throws Exception if (-put.equals(cmd) | -copyFromLocal.equals(cmd) Path srcs = new Pathargv.length-2; for (int j=0 ; i argv.length-1 ;) srcsj+ = new Path(argvi+); copyFromLocal(srcs, argvi+); org.apache.hadoop.fs. FsShell:void copyFromLocal(Path srcs, String dstf) throws IOException Path dstPath = new Path(dstf); FileSystem dstFs = dstPath.getFileSystem(getConf(); if (srcs.length = 1 & srcs0.toString().equals(-) copyFromStdin(dstPath, dstFs); else dstFs.copyFromLocalFile(false, false, srcs, dstPath); org.apache.hadoop.fs. FileSystem:public void copyFromLocalFile(boolean delSrc, boolean overwrite, Path srcs, Path dst) throws IOException Configuration conf = getConf(); FileUtil.copy(getLocal(conf), srcs, this, dst, delSrc, overwrite, conf); org.apache.hadoop.fs. FileUtil:public static boolean copy(FileSystem srcFS, Path srcs, FileSystem dstFS, Path dst, boolean deleteSource, boolean overwrite, Configuration conf) throws IOException if (srcs.length = 1) return copy(srcFS, srcs0, dstFS, dst, deleteSource, overwrite, conf);for (Path src : srcs) try if (!copy(srcFS, src, dstFS, dst, deleteSource, overwrite, conf) returnVal = false; catch (IOException e) gotException = true; exceptions.append(e.getMessage(); exceptions.append(n); return returnVal;在FsShell类中，定义了hadoop命令的解析，有run()函数检查hadoop shell命令的正确性并初始化shell，然后针对不同命令调用不同处理函数。本文是进入-put和-copyFromLocal的处理，接着调用copyFromLocal()函数；该函数会判断输入源是shell标准输入还是本地文件，本文进入本地文件，即调用copyFromLocalFile()函数，经过参数的完善，再调用FileUtil.copy()函数；该函数会判断输入文件是一个还是多个，我们为简单起见，按一个处理，从而进入另一个重载的copy()函数。org.apache.hadoop.fs. FileUtil:public static boolean copy(FileSystem srcFS, Path src, FileSystem dstFS, Path dst, boolean deleteSource, boolean overwrite, Configuration conf) throws IOException dst = checkDest(src.getName(), dstFS, dst, overwrite); if (srcFS.getFileStatus(src).isDir() checkDependencies(srcFS, src, dstFS, dst); if (!dstFS.mkdirs(dst) return false; FileStatus contents = srcFS.listStatus(src); for (int i = 0; i bytesPerChecksum) throw new IOException(writeChunk() buffer size is + len + is larger than supported bytesPerChecksum + bytesPerChecksum); if (checksum.length != this.checksum.getChecksumSize() throw new IOException(writeChunk() checksum size is supposed to be + this.checksum.getChecksumSize() + but found to be + checksum.length); synchronized (dataQueue) / If queue is full, then wait till we can create enough space while (!closed & dataQueue.size() + ackQueue.size() maxPackets) try dataQueue.wait(); catch (InterruptedException e) isClosed(); if (currentPacket = null) currentPacket = new Packet(packetSize, chunksPerPacket, bytesCurBlock); if (LOG.isDebugEnabled() LOG.debug(DFSClient writeChunk allocating new packet seqno= + currentPacket.seqno + , src= + src + , packetSize= + packetSize + , chunksPerPacket= + chunksPerPacket + , bytesCurBlock= + bytesCurBlock); currentPacket.writeChecksum(checksum, 0, cklen); currentPacket.writeData(b, offset, len); currentPacket.numChunks+; bytesCurBlock += len; / If packet is full, enqueue it for transmission / if (currentPacket.numChunks = currentPacket.maxChunks | bytesCurBlock = blockSize) if (LOG.isDebugEnabled() LOG.debug(DFSClient writeChunk packet full seqno= + currentPacket.seqno + , src= + src + , bytesCurBlock= + bytesCurBlock + , blockSize= + blockSize + , appendChunk= + appendChunk); / / if we allocated a new packet because we encountered a block / boundary, reset bytesCurBlock. / if (bytesCurBlock = blockSize) currentPacket.lastPacketInBlock = true; bytesCurBlock = 0; lastFlushOffset = 0; enqueueCurrentPacket(); / If this was the first write after reopening a file, then the above / write filled up any partial chunk. Tell the summer to generate full / crc chunks from now on. if (appendChunk) appendChunk = false; resetChecksumChunk(bytesPerChecksum); int psize = Math.min(int)(blockSize-bytesCurBlock), writePacketSize); computePacketChunkSize(psize, bytesPerChecksum); /LOG.debug(DFSClient writeChunk done length + len + / checksum length + cklen); org.apache.hadoop.hdfs.DFSClient.DFSOutputStreamprivate synchronized void enqueueCurrentPacket() synchronized (dataQueue) if (currentPacket = null) return; dataQueue.addLast(currentPacket); dataQueue.notifyAll(); lastQueuedSeqno = currentPacket.seqno; currentPacket = null; DFSOutputStream中维护了两个重要的队列：org.apache.hadoop.hdfs.DFSClient.DFSOutputStreamprivate LinkedList dataQueue = new LinkedList();private LinkedList ackQueue = new LinkedList();负责保证以上流水线持续进行。writeChunk()会先检查data queue是否已满，若满就等到有足够的空间来创建队列中下一个packet。若currentPacket为空，则创建一个new Packet，并将该packet写上checksum、数据、序号等信息；接着检查该packet是否已满，若满则放入data queue，并唤醒等待data queue的传输线程，即DataStreamer。org.apache.hadoop.hdfs.DFSClient.DFSOutputStream.DataStreamerpublic void run() long lastPacket = 0; while (!closed & clientRunning) / if the Responder encountered an error, shutdown Responder if (hasError & response != null) try response.close(); response.join(); response = null; catch (InterruptedException e) Packet one = null; synchronized (dataQueue) / process IO errors if any boolean doSleep = processDatanodeError(hasError, false); / wait for a packet to be sent. long now = System.currentTimeMillis(); while (!closed & !hasError & clientRunning & dataQueue.size() = 0 & (blockStream = null | ( blockStream !=