学步园

reduce执行流程经历三个阶段：copy、sort、reduce，在第一阶段reduce任务会把map的输出拷贝至本地，通过线程MapOutputCopier，该线程通过http协议将map输出拷贝至本地，该copy操作可以并行进行，默认情况下有5个线程执行此操作，如果map数量较大时可以适当调大此值，拷贝时使用http协议，此时reducetask为client，map端以jetty作为web服务器。reduce任务的执行与map一样在Child类启动，但在TaskFinal.run(job,umbilical)进入ReduceTask类执行。reduce的过程比较复杂，本节只分析copy部分，最后会分析整个reduce流程，需要注意的是每个reduce只拷贝自己需要处理那个partition数据。

拷贝map输出结果代码：ReduceTask.java 1309行

/** Loop forever and fetch map outputs as they become available.<br /> * The thread exits when it is interrupted by {@link ReduceTaskRunner}<br /> */<br /> @Override<br /> public void run() {<br /> while (true) {<br /> try {<br /> MapOutputLocation loc = null;<br /> long size = -1;<br /> //不停查询调度拷贝的集合，如果有数据到来，则说明需要做copy操作<br /> synchronized (scheduledCopies) {<br /> while (scheduledCopies.isEmpty()) {<br /> scheduledCopies.wait();<br /> }<br /> //取出第一个输出位置<br /> loc = scheduledCopies.remove(0);<br /> }<br /> CopyOutputErrorType error = CopyOutputErrorType.OTHER_ERROR;<br /> readError = false;<br /> try {<br /> shuffleClientMetrics.threadBusy();<br /> start(loc);<br /> //开始读取数据并返回长度，这也是下面代码中要分析的函数<br /> size = copyOutput(loc);<br /> //更新统计信息<br /> shuffleClientMetrics.successFetch();<br /> error = CopyOutputErrorType.NO_ERROR;<br /> } catch (IOException e) {<br /> LOG.warn(reduceTask.getTaskID() + " copy failed: " +<br /> loc.getTaskAttemptId() + " from " + loc.getHost());<br /> LOG.warn(StringUtils.stringifyException(e));<br /> shuffleClientMetrics.failedFetch();<br /> if (readError) {<br /> error = CopyOutputErrorType.READ_ERROR;<br /> }<br /> // Reset<br /> size = -1;<br /> } finally {<br /> shuffleClientMetrics.threadFree();<br /> //当前连接加入copyResults集合<br /> finish(size, error);<br /> }<br /> } catch (InterruptedException e) {<br /> break; // 执行到此步则证明拷贝完成<br /> } catch (FSError e) {//文件系统错误<br /> LOG.error("Task: " + reduceTask.getTaskID() + " - FSError: " +<br /> StringUtils.stringifyException(e));<br /> try {<br /> umbilical.fsError(reduceTask.getTaskID(), e.getMessage(), jvmContext);<br /> } catch (IOException io) {<br /> LOG.error("Could not notify TT of FSError: " +<br /> StringUtils.stringifyException(io));<br /> }<br /> } catch (Throwable th) {<br /> String msg = getTaskID() + " : Map output copy failure : "<br /> + StringUtils.stringifyException(th);<br /> reportFatalError(getTaskID(), th, msg);<br /> }<br /> }<br /> if (decompressor != null) {<br /> CodecPool.returnDecompressor(decompressor);<br /> }<br /> }<br />

下面分析copyOutput函数：ReduceTask.java 1373行

/** Copies a a map output from a remote host, via HTTP.<br /> * @param currentLocation the map output location to be copied<br /> * @return the path (fully qualified) of the copied file<br /> * @throws IOException if there is an error copying the file<br /> * @throws InterruptedException if the copier should give up<br /> */<br /> private long copyOutput(MapOutputLocation loc<br /> ) throws IOException, InterruptedException {<br /> // 检查该位置是否还需拷贝<br /> if (copiedMapOutputs.contains(loc.getTaskId()) ||<br /> obsoleteMapIds.contains(loc.getTaskAttemptId())) {<br /> return CopyResult.OBSOLETE;<br /> } </p> <p> // 内存写满时需要用到的临时文件<br /> TaskAttemptID reduceId = reduceTask.getTaskID();<br /> Path filename =<br /> new Path(String.format(<br /> MapOutputFile.REDUCE_INPUT_FILE_FORMAT_STRING,<br /> TaskTracker.OUTPUT, loc.getTaskId().getId()));</p> <p> // Copy the map output to a temp file whose name is unique to this attempt<br /> Path tmpMapOutput = new Path(filename+"-"+id);</p> <p> // 开始拷贝map的输出<br /> MapOutput mapOutput = getMapOutput(loc, tmpMapOutput,<br /> reduceId.getTaskID().getId());<br /> if (mapOutput == null) {<br /> throw new IOException("Failed to fetch map-output for " +<br /> loc.getTaskAttemptId() + " from " +<br /> loc.getHost());<br /> }</p> <p> // 获得输出尺寸<br /> long bytes = mapOutput.compressedSize;</p> <p> // lock the ReduceTask while we do the rename<br /> synchronized (ReduceTask.this) {<br /> if (copiedMapOutputs.contains(loc.getTaskId())) {<br /> mapOutput.discard();<br /> return CopyResult.OBSOLETE;<br /> }</p> <p> // Special case: discard empty map-outputs<br /> if (bytes == 0) {<br /> try {<br /> mapOutput.discard();<br /> } catch (IOException ioe) {<br /> LOG.info("Couldn't discard output of " + loc.getTaskId());<br /> }</p> <p> // Note that we successfully copied the map-output<br /> noteCopiedMapOutput(loc.getTaskId());</p> <p> return bytes;<br /> }</p> <p> // 判断是否完全在内存中，根据具体情况执行不同分支<br /> if (mapOutput.inMemory) {<br /> // 如果完全在内存中则放入内存文件的集合中<br /> mapOutputsFilesInMemory.add(mapOutput);<br /> } else {<br /> // Rename the temporary file to the final file;<br /> // ensure it is on the same partition<br /> tmpMapOutput = mapOutput.file;<br /> filename = new Path(tmpMapOutput.getParent(), filename.getName());<br /> if (!localFileSys.rename(tmpMapOutput, filename)) {<br /> localFileSys.delete(tmpMapOutput, true);<br /> bytes = -1;<br /> throw new IOException("Failed to rename map output " +<br /> tmpMapOutput + " to " + filename);<br /> }</p> <p> synchronized (mapOutputFilesOnDisk) {<br /> addToMapOutputFilesOnDisk(localFileSys.getFileStatus(filename));<br /> }<br /> }</p> <p> // Note that we successfully copied the map-output<br /> noteCopiedMapOutput(loc.getTaskId());<br /> }</p> <p> return bytes;<br /> }<br />

getMapOutput函数负责拷贝输出的工作，利用URLConnection建立连接，url格式类似：http://PC-20130917RGUY:50060/mapOutput?job=job_201311261309_0003&map=attempt_201311261309_0003_m_000000_0&reduce=1
，包含协议类型:http,主机及端口：PC-20130917RGUY:50060,路径名称：mapOutput，查询参数包含作业名、map任务名、reduce编号：ob=job_201311261309_0003&map=attempt_201311261309_0003_m_000000_0&reduce=1
url会根据这个地址建立连接，并打开一个输入流读取数据。在开始读取前会判断本次的读取是否能全部放入缓存中，这部分缓存使用是有限制的：jvm_heap_size × mapred.job.shuffle.input.buffer.percent × MAX_SINGLE_SHUFFLE_SEGMENT_FRACTION，其中jvm_heap_size可以通过mapred.job.reduce.total.mem.bytes来设置，如果没设置则通过Runtime.getRuntime().maxMemory()来获取，可以通过mapred.child.opts来影响jvm堆的大小，mapred.job.shuffle.input.buffer.percent可以在参数文件中设置，默认为0.7，MAX_SINGLE_SHUFFLE_SEGMENT_FRACTION在当前版本中为一常量值0.25，也就是说加入我们指定jvm堆大小为1024M，那么一个ReduceTask在拷贝时用到的缓存为1024×0.7×0.25=179M，当我们的map输出大于179M时，则直接写入文件.

ReduceTask.java 1373行

<br /> private MapOutput getMapOutput(MapOutputLocation mapOutputLoc,<br /> Path filename, int reduce)<br /> throws IOException, InterruptedException {<br /> // 建立url连接<br /> URL url = mapOutputLoc.getOutputLocation();<br /> URLConnection connection = url.openConnection();</p> <p> InputStream input = setupSecureConnection(mapOutputLoc, connection);</p> <p> // 校验任务ID<br /> TaskAttemptID mapId = null;<br /> try {<br /> mapId =<br /> TaskAttemptID.forName(connection.getHeaderField(FROM_MAP_TASK));<br /> } catch (IllegalArgumentException ia) {<br /> LOG.warn("Invalid map id ", ia);<br /> return null;<br /> }<br /> TaskAttemptID expectedMapId = mapOutputLoc.getTaskAttemptId();<br /> if (!mapId.equals(expectedMapId)) {<br /> LOG.warn("data from wrong map:" + mapId +<br /> " arrived to reduce task " + reduce +<br /> ", where as expected map output should be from " + expectedMapId);<br /> return null;<br /> }<br /> //判断返回数据长度是否异常<br /> //取得压缩和未压缩长度，后面判断在内存还是硬盘做shuffle<br /> long decompressedLength =<br /> Long.parseLong(connection.getHeaderField(RAW_MAP_OUTPUT_LENGTH));<br /> long compressedLength =<br /> Long.parseLong(connection.getHeaderField(MAP_OUTPUT_LENGTH));</p> <p> if (compressedLength < 0 || decompressedLength < 0) {<br /> LOG.warn(getName() + " invalid lengths in map output header: id: " +<br /> mapId + " compressed len: " + compressedLength +<br /> ", decompressed len: " + decompressedLength);<br /> return null;<br /> }<br /> //判断reduce编号是否相同<br /> int forReduce =<br /> (int)Integer.parseInt(connection.getHeaderField(FOR_REDUCE_TASK));</p> <p> if (forReduce != reduce) {<br /> LOG.warn("data for the wrong reduce: " + forReduce +<br /> " with compressed len: " + compressedLength +<br /> ", decompressed len: " + decompressedLength +<br /> " arrived to reduce task " + reduce);<br /> return null;<br /> }<br /> if (LOG.isDebugEnabled()) {<br /> LOG.debug("header: " + mapId + ", compressed len: " + compressedLength +<br /> ", decompressed len: " + decompressedLength);<br /> }</p> <p> //We will put a file in memory if it meets certain criteria:<br /> //1. The size of the (decompressed) file should be less than 25% of<br /> // the total inmem fs<br /> //2. There is space available in the inmem fs</p> <p> // 判断拷贝的数据能否完全放入内存中，内存计算公式为: JVM堆尺寸×mapred.job.shuffle.input.buffer.percent（0.7）× 1/4<br /> boolean shuffleInMemory = ramManager.canFitInMemory(decompressedLength); </p> <p> // Shuffle<br /> MapOutput mapOutput = null;<br /> if (shuffleInMemory) {<br /> if (LOG.isDebugEnabled()) {<br /> LOG.debug("Shuffling " + decompressedLength + " bytes (" +<br /> compressedLength + " raw bytes) " +<br /> "into RAM from " + mapOutputLoc.getTaskAttemptId());<br /> }<br /> //如果可以放入内存，则放入新建立的byte buffer中<br /> mapOutput = shuffleInMemory(mapOutputLoc, connection, input,<br /> (int)decompressedLength,<br /> (int)compressedLength);<br /> } else {<br /> if (LOG.isDebugEnabled()) {<br /> LOG.debug("Shuffling " + decompressedLength + " bytes (" +<br /> compressedLength + " raw bytes) " +<br /> "into Local-FS from " + mapOutputLoc.getTaskAttemptId());<br /> }<br /> //内存中放不下则放入磁盘中<br /> mapOutput = shuffleToDisk(mapOutputLoc, input, filename,<br /> compressedLength);<br /> }</p> <p> return mapOutput;<br /> }<br />

如果内存足够大，则copy过来的数据直接放入内存中，首先会分配一个byte数组，然后从上面建立的输入流冲取得所需数据。

ReduceTask.java 1646行

<br /> private MapOutput shuffleInMemory(MapOutputLocation mapOutputLoc,<br /> URLConnection connection,<br /> InputStream input,<br /> int mapOutputLength,<br /> int compressedLength)<br /> throws IOException, InterruptedException {<br /> // 判断是否有足够内存存放数据，如果没有则等待内存刷新，<br /> //刷新内存前会讲输入流置空，所以在这个函数返回为false时需要重新连接，刷新数据时会唤醒内存合并线程<br /> boolean createdNow = ramManager.reserve(mapOutputLength, input);</p> <p> //是否需要重新连接<br /> if (!createdNow) {<br /> // Reconnect<br /> try {<br /> connection = mapOutputLoc.getOutputLocation().openConnection();<br /> input = setupSecureConnection(mapOutputLoc, connection);<br /> } catch (IOException ioe) {<br /> LOG.info("Failed reopen connection to fetch map-output from " +<br /> mapOutputLoc.getHost());</p> <p> // Inform the ram-manager<br /> ramManager.closeInMemoryFile(mapOutputLength);<br /> ramManager.unreserve(mapOutputLength);</p> <p> throw ioe;<br /> }<br /> }<br /> //包装输入流<br /> IFileInputStream checksumIn =<br /> new IFileInputStream(input,compressedLength);</p> <p> input = checksumIn; </p> <p> // Are map-outputs compressed?<br /> if (codec != null) {<br /> decompressor.reset();<br /> input = codec.createInputStream(input, decompressor);<br /> }</p> <p> // 创建buffer，从输入流读取并填充<br /> byte[] shuffleData = new byte[mapOutputLength];<br /> MapOutput mapOutput =<br /> new MapOutput(mapOutputLoc.getTaskId(),<br /> mapOutputLoc.getTaskAttemptId(), shuffleData, compressedLength);</p> <p> int bytesRead = 0;<br /> try {<br /> //循环读取流冲数据至缓存中<br /> int n = input.read(shuffleData, 0, shuffleData.length);<br /> while (n > 0) {<br /> bytesRead += n;<br /> shuffleClientMetrics.inputBytes(n);</p> <p> // indicate we're making progress<br /> reporter.progress();<br /> n = input.read(shuffleData, bytesRead,<br /> (shuffleData.length-bytesRead));<br /> }</p> <p> if (LOG.isDebugEnabled()) {<br /> LOG.debug("Read " + bytesRead + " bytes from map-output for " +<br /> mapOutputLoc.getTaskAttemptId());<br /> }<br /> //数据读取完毕则关闭输入流<br /> input.close();<br /> } catch (IOException ioe) {<br /> LOG.info("Failed to shuffle from " + mapOutputLoc.getTaskAttemptId(),<br /> ioe);</p> <p> .....<br /> }</p> <p> // 关闭内存文件，并唤醒内存合并线程<br /> ramManager.closeInMemoryFile(mapOutputLength);</p> <p> // 校验数据读取长度<br /> if (bytesRead != mapOutputLength) {<br /> // Inform the ram-manager<br /> ramManager.unreserve(mapOutputLength);</p> <p> // Discard the map-output<br /> try {<br /> mapOutput.discard();<br /> } catch (IOException ignored) {<br /> // IGNORED because we are cleaning up<br /> LOG.info("Failed to discard map-output from " +<br /> mapOutputLoc.getTaskAttemptId(), ignored);<br /> }<br /> mapOutput = null;</p> <p> throw new IOException("Incomplete map output received for " +<br /> mapOutputLoc.getTaskAttemptId() + " from " +<br /> mapOutputLoc.getOutputLocation() + " (" +<br /> bytesRead + " instead of " +<br /> mapOutputLength + ")"<br /> );<br /> }<br /> .....<br /> return mapOutput;<br /> }<br />

如果内存过小不能存放本次读取的数据则直接写入磁盘文件中，我们会在相关目录中看到这个文件如：C:/hadoop/tasklog/taskTracker/hadoop/jobcache/job_201311281345_0001/attempt_201311281345_0001_r_000001_1/output/map_0.out-0

private MapOutput shuffleToDisk(MapOutputLocation mapOutputLoc,<br /> InputStream input,<br /> Path filename,<br /> long mapOutputLength)<br /> throws IOException {<br /> // 构建本地文件系统文件名<br /> Path localFilename =<br /> lDirAlloc.getLocalPathForWrite(filename.toUri().getPath(),<br /> mapOutputLength, conf);<br /> //创建基于磁盘文件的MapOutput<br /> MapOutput mapOutput =<br /> new MapOutput(mapOutputLoc.getTaskId(), mapOutputLoc.getTaskAttemptId(),<br /> conf, localFileSys.makeQualified(localFilename),<br /> mapOutputLength);</p> <p> // 开始数据拷贝<br /> OutputStream output = null;<br /> long bytesRead = 0;<br /> try {<br /> output = rfs.create(localFilename);<br /> //讲map输出直接写入磁盘时分配的缓存，固定64K<br /> byte[] buf = new byte[64 * 1024];<br /> int n = -1;<br /> try {<br /> n = input.read(buf, 0, buf.length);<br /> } catch (IOException ioe) {<br /> readError = true;<br /> throw ioe;<br /> }<br /> while (n > 0) {<br /> bytesRead += n;<br /> shuffleClientMetrics.inputBytes(n);<br /> output.write(buf, 0, n);</p> <p> // indicate we're making progress<br /> reporter.progress();<br /> try {<br /> n = input.read(buf, 0, buf.length);<br /> } catch (IOException ioe) {<br /> readError = true;<br /> throw ioe;<br /> }<br /> }</p> <p> LOG.info("Read " + bytesRead + " bytes from map-output for " +<br /> mapOutputLoc.getTaskAttemptId());</p> <p> output.close();<br /> input.close();<br /> } catch (IOException ioe) {<br /> LOG.info("Failed to shuffle from " + mapOutputLoc.getTaskAttemptId(),<br /> ioe);</p> <p> // Discard the map-output<br /> try {<br /> mapOutput.discard();<br /> } catch (IOException ignored) {<br /> LOG.info("Failed to discard map-output from " +<br /> mapOutputLoc.getTaskAttemptId(), ignored);<br /> }<br /> mapOutput = null;</p> <p> // Close the streams<br /> IOUtils.cleanup(LOG, input, output);</p> <p> // Re-throw<br /> throw ioe;<br /> }</p> <p> // 读取数据后的检测<br /> if (bytesRead != mapOutputLength) {<br /> try {<br /> mapOutput.discard();<br /> } catch (Exception ioe) {<br /> // IGNORED because we are cleaning up<br /> LOG.info("Failed to discard map-output from " +<br /> mapOutputLoc.getTaskAttemptId(), ioe);<br /> } catch (Throwable t) {<br /> String msg = getTaskID() + " : Failed in shuffle to disk :"<br /> + StringUtils.stringifyException(t);<br /> reportFatalError(getTaskID(), t, msg);<br /> }<br /> mapOutput = null;</p> <p> throw new IOException("Incomplete map output received for " +<br /> mapOutputLoc.getTaskAttemptId() + " from " +<br /> mapOutputLoc.getOutputLocation() + " (" +<br /> bytesRead + " instead of " +<br /> mapOutputLength + ")"<br /> );<br /> }</p> <p> return mapOutput;<br /> }<br />

阅读笔记：

1. MapOutput类 本质上是一个指向一个数据块的指针，该数据块可以在硬盘上，也可以在内存上。(1)final boolean inMemory表示该数据块是否在内存中 (2)final Path file表示数据在硬盘上的路径 (3)byte[] data表示数据在内存中的数据块

2. ReduceTask.run() 是ReduceTask的起始点。其中分为三部分：(1).Copy阶段（由reduceCopier.fetchOutputs()完成）
(2).Sort阶段（由Merger.merge()完成） (3).Reduce阶段（由runOldReducer()或者runNewReducer()完成）

3. fetchOutputs()函数中启动多个（由mapred.reduce.parallel.copies属性设置，默认为5个）MapOutputCopier线程进行远程数据拷贝到本地。远程拷贝运行过程中，存在 InMemFSMergeThread线程 和 LocalFSMerger线程 进行文件合并。

4. MapOutput类中的 discard()函数即抛弃拷贝的map输出结果。若该MapOutput数据块在内存上，则将数据指针data置null，若数据块在硬盘上，则调用 fs.delete(file,true) 删除该文件。

5. 远程拷贝过程中，每次拷贝一个数据块时，若该数据块可以放入内存则放入内存，否则放入硬盘。有两个标准决定该数据块是否应该放入硬盘：(1) 数据块小于 java_heaps _size * mapred.job.shuffle.input.buffer.percent * MAX_SINGLE_SHUFFLE_SEGMENT_FRACTION(0.25) (2) 内存中有足够空间放入该数据块。