public void run() {
        while (true) {        
            MapOutputLocation loc = null;
            long size = -1;
              synchronized (scheduledCopies) {
              while (scheduledCopies.isEmpty()) {
                scheduledCopies.wait();
              }
              loc = scheduledCopies.remove(0);
            }            
                     
              start(loc);
              size = copyOutput(loc);
             
        
        if (decompressor != null) {
          CodecPool.returnDecompressor(decompressor);
        }
          
      }

private long copyOutput(MapOutputLocation loc
            ) throws IOException, InterruptedException {
        // 从拷贝的记录中检查是否已经拷贝完成。
        if (copiedMapOutputs.contains(loc.getTaskId()) || 
                obsoleteMapIds.contains(loc.getTaskAttemptId())) {
            return CopyResult.OBSOLETE;
        } 
        TaskAttemptID reduceId = reduceTask.getTaskID();
        Path filename = new Path("/" + TaskTracker.getIntermediateOutputDir(
                reduceId.getJobID().toString(),
                reduceId.toString()) 
                + "/map_" +
                loc.getTaskId().getId() + ".out");

        //一个拷贝map输出的临时文件。
        Path tmpMapOutput = new Path(filename+"-"+id);

        //拷贝map输出。
        MapOutput mapOutput = getMapOutput(loc, tmpMapOutput);
        if (mapOutput == null) {
            throw new IOException("Failed to fetch map-output for " + 
                    loc.getTaskAttemptId() + " from " + 
                    loc.getHost());
        }
        // The size of the map-output
        long bytes = mapOutput.compressedSize;

        synchronized (ReduceTask.this) {
            if (copiedMapOutputs.contains(loc.getTaskId())) {
                mapOutput.discard();
                return CopyResult.OBSOLETE;
            }
            // Note that we successfully copied the map-output
            noteCopiedMapOutput(loc.getTaskId());
            return bytes;
        }

        // 处理map的输出，如果是存储在内存中则添加到（Collections.synchronizedList(new LinkedList<MapOutput>)类型的结合mapOutputsFilesInMemory中，否则如果存储在临时文件中，则冲明明临时文件为正式的输出文件。
        if (mapOutput.inMemory) {
            // Save it in the synchronized list of map-outputs
            mapOutputsFilesInMemory.add(mapOutput);
        } else {

            tmpMapOutput = mapOutput.file;
            filename = new Path(tmpMapOutput.getParent(), filename.getName());
            if (!localFileSys.rename(tmpMapOutput, filename)) {
                localFileSys.delete(tmpMapOutput, true);
                bytes = -1;
                throw new IOException("Failed to rename map output " + 
                        tmpMapOutput + " to " + filename);
            }

            synchronized (mapOutputFilesOnDisk) {        
                addToMapOutputFilesOnDisk(localFileSys.getFileStatus(filename));
            }
        }

        // Note that we successfully copied the map-output
        noteCopiedMapOutput(loc.getTaskId());
    }

    return bytes;
}

private MapOutput getMapOutput(MapOutputLocation mapOutputLoc, 
            Path filename, int reduce)
                    throws IOException, InterruptedException {
        // 根据远端服务器地址构建连接。
        URLConnection connection = 
                mapOutputLoc.getOutputLocation().openConnection();
        InputStream input = getInputStream(connection, STALLED_COPY_TIMEOUT,
                DEFAULT_READ_TIMEOUT); 

        // 从输出的http header中得到mapid
        TaskAttemptID mapId = null;
        mapId =           TaskAttemptID.forName(connection.getHeaderField(FROM_MAP_TASK));

        TaskAttemptID expectedMapId = mapOutputLoc.getTaskAttemptId();
        if (!mapId.equals(expectedMapId)) {
            LOG.warn("data from wrong map:" + mapId +
                    " arrived to reduce task " + reduce +
                    ", where as expected map output should be from " + expectedMapId);
            return null;
        }

        long decompressedLength = 
                Long.parseLong(connection.getHeaderField(RAW_MAP_OUTPUT_LENGTH));  
        long compressedLength = 
                Long.parseLong(connection.getHeaderField(MAP_OUTPUT_LENGTH));

        if (compressedLength < 0 || decompressedLength < 0) {
            LOG.warn(getName() + " invalid lengths in map output header: id: " +
                    mapId + " compressed len: " + compressedLength +
                    ", decompressed len: " + decompressedLength);
            return null;
        }
        int forReduce =
                (int)Integer.parseInt(connection.getHeaderField(FOR_REDUCE_TASK));

        if (forReduce != reduce) {
            LOG.warn("data for the wrong reduce: " + forReduce +
                    " with compressed len: " + compressedLength +
                    ", decompressed len: " + decompressedLength +
                    " arrived to reduce task " + reduce);
            return null;
        }
        LOG.info("header: " + mapId + ", compressed len: " + compressedLength +
                ", decompressed len: " + decompressedLength);


        // 检查map的输出大小是否能在memory里存储下，已决定是在内存中shuffle还是在磁盘上shuffle。并决定最终生成的MapOutput对象调用不同的构造函数，其inMemory属性页不同。
        boolean shuffleInMemory = ramManager.canFitInMemory(decompressedLength); 

        // Shuffle
        MapOutput mapOutput = null;
        if (shuffleInMemory) { 
            LOG.info("Shuffling " + decompressedLength + " bytes (" + 
                    compressedLength + " raw bytes) " + 
                    "into RAM from " + mapOutputLoc.getTaskAttemptId());

            mapOutput = shuffleInMemory(mapOutputLoc, connection, input,
                    (int)decompressedLength,
                    (int)compressedLength);
        } else {
            LOG.info("Shuffling " + decompressedLength + " bytes (" + 
                    compressedLength + " raw bytes) " + 
                    "into Local-FS from " + mapOutputLoc.getTaskAttemptId());

            mapOutput = shuffleToDisk(mapOutputLoc, input, filename, 
                    compressedLength);
        }

        return mapOutput;
    }

private MapOutput shuffleInMemory(MapOutputLocation mapOutputLoc,
            URLConnection connection, 
            InputStream input,
            int mapOutputLength,
            int compressedLength)
                    throws IOException, InterruptedException {

        //checksum 输入流，读Mpareduce中间文件IFile.
        IFileInputStream checksumIn = 
                new IFileInputStream(input,compressedLength);

        input = checksumIn;       

        // 如果加密，则根据codec来构建一个解密的输入流。
        if (codec != null) {
            decompressor.reset();
            input = codec.createInputStream(input, decompressor);
        }

        //把map的输出拷贝到内存的buffer中。
        byte[] shuffleData = new byte[mapOutputLength];
        MapOutput mapOutput = 
                new MapOutput(mapOutputLoc.getTaskId(), 
                        mapOutputLoc.getTaskAttemptId(), shuffleData, compressedLength);

        int bytesRead = 0;
        try {
            int n = input.read(shuffleData, 0, shuffleData.length);
            while (n > 0) {
                bytesRead += n;
                shuffleClientMetrics.inputBytes(n);

                // indicate we‘re making progress
                reporter.progress();
                n = input.read(shuffleData, bytesRead, 
                        (shuffleData.length-bytesRead));
            }

            LOG.info("Read " + bytesRead + " bytes from map-output for " +
                    mapOutputLoc.getTaskAttemptId());

            input.close();
        } catch (IOException ioe) {
            LOG.info("Failed to shuffle from " + mapOutputLoc.getTaskAttemptId(), 
                    ioe);

            // Inform the ram-manager
            ramManager.closeInMemoryFile(mapOutputLength);
            ramManager.unreserve(mapOutputLength);

            // Discard the map-output
            try {
                mapOutput.discard();
            } catch (IOException ignored) {
                LOG.info("Failed to discard map-output from " + 
                        mapOutputLoc.getTaskAttemptId(), ignored);
            }
            mapOutput = null;

            // Close the streams
            IOUtils.cleanup(LOG, input);

            // Re-throw
            throw ioe;
        }

        // Close the in-memory file
        ramManager.closeInMemoryFile(mapOutputLength);

        // Sanity check
        if (bytesRead != mapOutputLength) {
            // Inform the ram-manager
            ramManager.unreserve(mapOutputLength);

            // Discard the map-output
            try {
                mapOutput.discard();
            } catch (IOException ignored) {
                // IGNORED because we are cleaning up
                LOG.info("Failed to discard map-output from " + 
                        mapOutputLoc.getTaskAttemptId(), ignored);
            }
            mapOutput = null;

            throw new IOException("Incomplete map output received for " +
                    mapOutputLoc.getTaskAttemptId() + " from " +
                    mapOutputLoc.getOutputLocation() + " (" + 
                    bytesRead + " instead of " + 
                    mapOutputLength + ")"
                    );
        }

        // TODO: Remove this after a ‘fix‘ for HADOOP-3647
        if (mapOutputLength > 0) {
            DataInputBuffer dib = new DataInputBuffer();
            dib.reset(shuffleData, 0, shuffleData.length);
            LOG.info("Rec #1 from " + mapOutputLoc.getTaskAttemptId() + " -> (" + 
                    WritableUtils.readVInt(dib) + ", " + 
                    WritableUtils.readVInt(dib) + ") from " + 
                    mapOutputLoc.getHost());
        }

        return mapOutput;
    }

private MapOutput shuffleToDisk(MapOutputLocation mapOutputLoc,
                                      InputStream input,
                                      Path filename,
                                      long mapOutputLength) 
      throws IOException {
        // Find out a suitable location for the output on local-filesystem
        Path localFilename = 
          lDirAlloc.getLocalPathForWrite(filename.toUri().getPath(), 
                                         mapOutputLength, conf);

        MapOutput mapOutput = 
          new MapOutput(mapOutputLoc.getTaskId(), mapOutputLoc.getTaskAttemptId(), 
                        conf, localFileSys.makeQualified(localFilename), 
                        mapOutputLength);


        // Copy data to local-disk
        OutputStream output = null;
        long bytesRead = 0;
        try {
          output = rfs.create(localFilename);
          
          byte[] buf = new byte[64 * 1024];
          int n = input.read(buf, 0, buf.length);
          while (n > 0) {
            bytesRead += n;
            shuffleClientMetrics.inputBytes(n);
            output.write(buf, 0, n);

            // indicate we‘re making progress
            reporter.progress();
            n = input.read(buf, 0, buf.length);
          }

          LOG.info("Read " + bytesRead + " bytes from map-output for " +
              mapOutputLoc.getTaskAttemptId());

          output.close();
          input.close();
        } catch (IOException ioe) {
          LOG.info("Failed to shuffle from " + mapOutputLoc.getTaskAttemptId(), 
                   ioe);

          // Discard the map-output
          try {
            mapOutput.discard();
          } catch (IOException ignored) {
            LOG.info("Failed to discard map-output from " + 
                mapOutputLoc.getTaskAttemptId(), ignored);
          }
          mapOutput = null;

          // Close the streams
          IOUtils.cleanup(LOG, input, output);

          // Re-throw
          throw ioe;
        }

        // Sanity check
        if (bytesRead != mapOutputLength) {
          try {
            mapOutput.discard();
          } catch (Exception ioe) {
            // IGNORED because we are cleaning up
            LOG.info("Failed to discard map-output from " + 
                mapOutputLoc.getTaskAttemptId(), ioe);
          } catch (Throwable t) {
            String msg = getTaskID() + " : Failed in shuffle to disk :" 
                         + StringUtils.stringifyException(t);
            reportFatalError(getTaskID(), t, msg);
          }
          mapOutput = null;

          throw new IOException("Incomplete map output received for " +
                                mapOutputLoc.getTaskAttemptId() + " from " +
                                mapOutputLoc.getOutputLocation() + " (" + 
                                bytesRead + " instead of " + 
                                mapOutputLength + ")"
          );
        }

        return mapOutput;

      }

public void run() {
        try {
            LOG.info(reduceTask.getTaskID() + " Thread started: " + getName());
            while(!exitLocalFSMerge){
                // TreeSet<FileStatus>(mapOutputFileComparator)中存储了mapout的本地文件集合。
                synchronized (mapOutputFilesOnDisk) {             
                    List<Path> mapFiles = new ArrayList<Path>();
                    long approxOutputSize = 0;
                    int bytesPerSum = 
                            reduceTask.getConf().getInt("io.bytes.per.checksum", 512);
                    LOG.info(reduceTask.getTaskID() + "We have  " + 
                            mapOutputFilesOnDisk.size() + " map outputs on disk. " +
                            "Triggering merge of " + ioSortFactor + " files");
                    // 1. Prepare the list of files to be merged. This list is prepared
                    // using a list of map output files on disk. Currently we merge
                    // io.sort.factor files into 1.
                    //1. io.sort.factor构造List<Path> mapFiles，准备合并。            synchronized (mapOutputFilesOnDisk) {
                    for (int i = 0; i < ioSortFactor; ++i) {
                        FileStatus filestatus = mapOutputFilesOnDisk.first();
                        mapOutputFilesOnDisk.remove(filestatus);
                        mapFiles.add(filestatus.getPath());
                        approxOutputSize += filestatus.getLen();
                    }
                }



                // add the checksum length
                approxOutputSize += ChecksumFileSystem
                        .getChecksumLength(approxOutputSize,
                                bytesPerSum);

                // 2. 对list中的文件进行合并。
                Path outputPath = 
                        lDirAlloc.getLocalPathForWrite(mapFiles.get(0).toString(), 
                                approxOutputSize, conf)
                                .suffix(".merged");
                Writer writer = 
                        new Writer(conf,rfs, outputPath, 
                                conf.getMapOutputKeyClass(), 
                                conf.getMapOutputValueClass(),
                                codec, null);
                RawKeyValueIterator iter  = null;
                Path tmpDir = new Path(reduceTask.getTaskID().toString());
                try {
                    iter = Merger.merge(conf, rfs,
                            conf.getMapOutputKeyClass(),
                            conf.getMapOutputValueClass(),
                            codec, mapFiles.toArray(new Path[mapFiles.size()]), 
                            true, ioSortFactor, tmpDir, 
                            conf.getOutputKeyComparator(), reporter,
                            spilledRecordsCounter, null);

                    Merger.writeFile(iter, writer, reporter, conf);
                    writer.close();
                } catch (Exception e) {
                    localFileSys.delete(outputPath, true);
                    throw new IOException (StringUtils.stringifyException(e));
                }

                synchronized (mapOutputFilesOnDisk) {
                    addToMapOutputFilesOnDisk(localFileSys.getFileStatus(outputPath));
                }
                LOG.info(reduceTask.getTaskID() +
                        " Finished merging " + mapFiles.size() + 
                        " map output files on disk of total-size " + 
                        approxOutputSize + "." + 
                        " Local output file is " + outputPath + " of size " +
                        localFileSys.getFileStatus(outputPath).getLen());
            }
        } catch (Throwable t) {
            LOG.warn(reduceTask.getTaskID()
                    + " Merging of the local FS files threw an exception: "
                    + StringUtils.stringifyException(t));
            if (mergeThrowable == null) {
                mergeThrowable = t;
            }
        } 
    }
}

    public void run() {
        LOG.info(reduceTask.getTaskID() + " Thread started: " + getName());
        try {
          boolean exit = false;
          do {
            exit = ramManager.waitForDataToMerge();
            if (!exit) {
              doInMemMerge();
            }
          } while (!exit);
        } catch (Throwable t) {
          LOG.warn(reduceTask.getTaskID() +
                   " Merge of the inmemory files threw an exception: "
                   + StringUtils.stringifyException(t));
          ReduceCopier.this.mergeThrowable = t;
        }
      }

private void doInMemMerge() throws IOException{
        if (mapOutputsFilesInMemory.size() == 0) {
          return;
        }
        
        TaskID mapId = mapOutputsFilesInMemory.get(0).mapId;

        List<Segment<K, V>> inMemorySegments = new ArrayList<Segment<K,V>>();
        long mergeOutputSize = createInMemorySegments(inMemorySegments, 0);
        int noInMemorySegments = inMemorySegments.size();

        Path outputPath = mapOutputFile.getInputFileForWrite(mapId, 
                          reduceTask.getTaskID(), mergeOutputSize);

        Writer writer = 
          new Writer(conf, rfs, outputPath,
                     conf.getMapOutputKeyClass(),
                     conf.getMapOutputValueClass(),
                     codec, null);

        RawKeyValueIterator rIter = null;
        try {
          LOG.info("Initiating in-memory merge with " + noInMemorySegments + 
                   " segments...");
          
          rIter = Merger.merge(conf, rfs,
                               (Class<K>)conf.getMapOutputKeyClass(),
                               (Class<V>)conf.getMapOutputValueClass(),
                               inMemorySegments, inMemorySegments.size(),
                               new Path(reduceTask.getTaskID().toString()),
                               conf.getOutputKeyComparator(), reporter,
                               spilledRecordsCounter, null);
          
          if (combinerRunner == null) {
            Merger.writeFile(rIter, writer, reporter, conf);
          } else {
            combineCollector.setWriter(writer);
            combinerRunner.combine(rIter, combineCollector);
          }
          writer.close();

          LOG.info(reduceTask.getTaskID() + 
              " Merge of the " + noInMemorySegments +
              " files in-memory complete." +
              " Local file is " + outputPath + " of size " + 
              localFileSys.getFileStatus(outputPath).getLen());
        } catch (Exception e) { 
          //make sure that we delete the ondisk file that we created 
          //earlier when we invoked cloneFileAttributes
          localFileSys.delete(outputPath, true);
          throw (IOException)new IOException
                  ("Intermediate merge failed").initCause(e);
        }

        // Note the output of the merge
        FileStatus status = localFileSys.getFileStatus(outputPath);
        synchronized (mapOutputFilesOnDisk) {
          addToMapOutputFilesOnDisk(status);
        }
      }
    }

public void run() {
      
        LOG.info(reduceTask.getTaskID() + " Thread started: " + getName());
        
        do {
          try {
            int numNewMaps = getMapCompletionEvents();
            if (numNewMaps > 0) {
              LOG.info(reduceTask.getTaskID() + ": " +  
                  "Got " + numNewMaps + " new map-outputs"); 
            }
            Thread.sleep(SLEEP_TIME);
          } 
          catch (InterruptedException e) {
            LOG.warn(reduceTask.getTaskID() +
                " GetMapEventsThread returning after an " +
                " interrupted exception");
            return;
          }
          catch (Throwable t) {
            LOG.warn(reduceTask.getTaskID() +
                " GetMapEventsThread Ignoring exception : " +
                StringUtils.stringifyException(t));
          }
        } while (!exitGetMapEvents);

        LOG.info("GetMapEventsThread exiting");
      
      }

    private int getMapCompletionEvents() throws IOException {

        int numNewMaps = 0;

        //RPC调用Tasktracker的getMapCompletionEvents方法，获得MapTaskCompletionEventsUpdate，进而获得TaskCompletionEvents
        MapTaskCompletionEventsUpdate update = 
                umbilical.getMapCompletionEvents(reduceTask.getJobID(), 
                        fromEventId.get(), 
                        MAX_EVENTS_TO_FETCH,
                        reduceTask.getTaskID());
        TaskCompletionEvent events[] = update.getMapTaskCompletionEvents();

        // Check if the reset is required.
        // Since there is no ordering of the task completion events at the 
        // reducer, the only option to sync with the new jobtracker is to reset 
        // the events index
        if (update.shouldReset()) {
            fromEventId.set(0);
            obsoleteMapIds.clear(); // clear the obsolete map
            mapLocations.clear(); // clear the map locations mapping
        }

        // Update the last seen event ID
        fromEventId.set(fromEventId.get() + events.length);

        // Process the TaskCompletionEvents:
        // 1. Save the SUCCEEDED maps in knownOutputs to fetch the outputs.
        // 2. Save the OBSOLETE/FAILED/KILLED maps in obsoleteOutputs to stop 
        //    fetching from those maps.
        // 3. Remove TIPFAILED maps from neededOutputs since we don‘t need their
        //    outputs at all.
        //对每个完成的Event，获取maptask所在的服务器地址，构造URL，加入到mapLocations，供copier线程获取。
        for (TaskCompletionEvent event : events) {
            switch (event.getTaskStatus()) {
            case SUCCEEDED:
            {
                URI u = URI.create(event.getTaskTrackerHttp());
                String host = u.getHost();
                TaskAttemptID taskId = event.getTaskAttemptId();
                int duration = event.getTaskRunTime();
                if (duration > maxMapRuntime) {
                    maxMapRuntime = duration; 
                    // adjust max-fetch-retries based on max-map-run-time
                    maxFetchRetriesPerMap = Math.max(MIN_FETCH_RETRIES_PER_MAP, 
                            getClosestPowerOf2((maxMapRuntime / BACKOFF_INIT) + 1));
                }
                URL mapOutputLocation = new URL(event.getTaskTrackerHttp() + 
                        "/mapOutput?job=" + taskId.getJobID() +
                        "&map=" + taskId + 
                        "&reduce=" + getPartition());
                List<MapOutputLocation> loc = mapLocations.get(host);
                if (loc == null) {
                    loc = Collections.synchronizedList
                            (new LinkedList<MapOutputLocation>());
                    mapLocations.put(host, loc);
                }
                loc.add(new MapOutputLocation(taskId, host, mapOutputLocation));
                numNewMaps ++;
            }
            break;
            case FAILED:
            case KILLED:
            case OBSOLETE:
            {
                obsoleteMapIds.add(event.getTaskAttemptId());
                LOG.info("Ignoring obsolete output of " + event.getTaskStatus() + 
                        " map-task: ‘" + event.getTaskAttemptId() + "‘");
            }
            break;
            case TIPFAILED:
            {
                copiedMapOutputs.add(event.getTaskAttemptId().getTaskID());
                LOG.info("Ignoring output of failed map TIP: ‘" +  
                        event.getTaskAttemptId() + "‘");
            }
            break;
            }
        }
        return numNewMaps;
    }
}
}

private RawKeyValueIterator createKVIterator(
        JobConf job, FileSystem fs, Reporter reporter) throws IOException {

      // merge config params
      Class<K> keyClass = (Class<K>)job.getMapOutputKeyClass();
      Class<V> valueClass = (Class<V>)job.getMapOutputValueClass();
      boolean keepInputs = job.getKeepFailedTaskFiles();
      final Path tmpDir = new Path(getTaskID().toString());
      final RawComparator<K> comparator =
        (RawComparator<K>)job.getOutputKeyComparator();

      // segments required to vacate memory
      List<Segment<K,V>> memDiskSegments = new ArrayList<Segment<K,V>>();
      long inMemToDiskBytes = 0;
      if (mapOutputsFilesInMemory.size() > 0) {
        TaskID mapId = mapOutputsFilesInMemory.get(0).mapId;
        inMemToDiskBytes = createInMemorySegments(memDiskSegments,
            maxInMemReduce);
        final int numMemDiskSegments = memDiskSegments.size();
        if (numMemDiskSegments > 0 &&
              ioSortFactor > mapOutputFilesOnDisk.size()) {
          // must spill to disk, but can‘t retain in-mem for intermediate merge
          final Path outputPath = mapOutputFile.getInputFileForWrite(mapId,
                            reduceTask.getTaskID(), inMemToDiskBytes);
          final RawKeyValueIterator rIter = Merger.merge(job, fs,
              keyClass, valueClass, memDiskSegments, numMemDiskSegments,
              tmpDir, comparator, reporter, spilledRecordsCounter, null);
          final Writer writer = new Writer(job, fs, outputPath,
              keyClass, valueClass, codec, null);
          try {
            Merger.writeFile(rIter, writer, reporter, job);
            addToMapOutputFilesOnDisk(fs.getFileStatus(outputPath));
          } catch (Exception e) {
            if (null != outputPath) {
              fs.delete(outputPath, true);
            }
            throw new IOException("Final merge failed", e);
          } finally {
            if (null != writer) {
              writer.close();
            }
          }
          LOG.info("Merged " + numMemDiskSegments + " segments, " +
                   inMemToDiskBytes + " bytes to disk to satisfy " +
                   "reduce memory limit");
          inMemToDiskBytes = 0;
          memDiskSegments.clear();
        } else if (inMemToDiskBytes != 0) {
          LOG.info("Keeping " + numMemDiskSegments + " segments, " +
                   inMemToDiskBytes + " bytes in memory for " +
                   "intermediate, on-disk merge");
        }
      }

      // segments on disk
      List<Segment<K,V>> diskSegments = new ArrayList<Segment<K,V>>();
      long onDiskBytes = inMemToDiskBytes;
      Path[] onDisk = getMapFiles(fs, false);
      for (Path file : onDisk) {
        onDiskBytes += fs.getFileStatus(file).getLen();
        diskSegments.add(new Segment<K, V>(job, fs, file, codec, keepInputs));
      }
      LOG.info("Merging " + onDisk.length + " files, " +
               onDiskBytes + " bytes from disk");
      Collections.sort(diskSegments, new Comparator<Segment<K,V>>() {
        public int compare(Segment<K, V> o1, Segment<K, V> o2) {
          if (o1.getLength() == o2.getLength()) {
            return 0;
          }
          return o1.getLength() < o2.getLength() ? -1 : 1;
        }
      });

      // build final list of segments from merged backed by disk + in-mem
      List<Segment<K,V>> finalSegments = new ArrayList<Segment<K,V>>();
      long inMemBytes = createInMemorySegments(finalSegments, 0);
      LOG.info("Merging " + finalSegments.size() + " segments, " +
               inMemBytes + " bytes from memory into reduce");
      if (0 != onDiskBytes) {
        final int numInMemSegments = memDiskSegments.size();
        diskSegments.addAll(0, memDiskSegments);
        memDiskSegments.clear();
        RawKeyValueIterator diskMerge = Merger.merge(
            job, fs, keyClass, valueClass, diskSegments,
            ioSortFactor, numInMemSegments, tmpDir, comparator,
            reporter, false, spilledRecordsCounter, null);
        diskSegments.clear();
        if (0 == finalSegments.size()) {
          return diskMerge;
        }
        finalSegments.add(new Segment<K,V>(
              new RawKVIteratorReader(diskMerge, onDiskBytes), true));
      }
      return Merger.merge(job, fs, keyClass, valueClass,
                   finalSegments, finalSegments.size(), tmpDir,
                   comparator, reporter, spilledRecordsCounter, null);
    }

@SuppressWarnings("unchecked")
    private <INKEY,INVALUE,OUTKEY,OUTVALUE>
    void runNewReducer(JobConf job,
            final TaskUmbilicalProtocol umbilical,
            final TaskReporter reporter,
            RawKeyValueIterator rIter,
            RawComparator<INKEY> comparator,
            Class<INKEY> keyClass,
            Class<INVALUE> valueClass
            ) throws IOException,InterruptedException, 
            ClassNotFoundException {
        //1. 构造TaskContext
        org.apache.hadoop.mapreduce.TaskAttemptContext taskContext =
                new org.apache.hadoop.mapreduce.TaskAttemptContext(job, getTaskID());
        //2. 根据配置的Reducer类构造一个Reducer实例
        org.apache.hadoop.mapreduce.Reducer<INKEY,INVALUE,OUTKEY,OUTVALUE> reducer =      (org.apache.hadoop.mapreduce.Reducer<INKEY,INVALUE,OUTKEY,OUTVALUE>)
                ReflectionUtils.newInstance(taskContext.getReducerClass(), job);
        //3. 构造RecordWriter
        org.apache.hadoop.mapreduce.RecordWriter<OUTKEY,OUTVALUE> output =
                (org.apache.hadoop.mapreduce.RecordWriter<OUTKEY,OUTVALUE>)
                outputFormat.getRecordWriter(taskContext);
        job.setBoolean("mapred.skip.on", isSkipping());

        //4. 构造Context，是Reducer运行的上下文
        org.apache.hadoop.mapreduce.Reducer.Context 
        reducerContext = createReduceContext(reducer, job, getTaskID(),
                rIter, reduceInputValueCounter, 
                output, committer,
                reporter, comparator, keyClass,
                valueClass);
        reducer.run(reducerContext);
        output.close(reducerContext);
    }

public void run(Context context) throws IOException, InterruptedException {
    setup(context);
    while (context.nextKey()) {
      reduce(context.getCurrentKey(), context.getValues(), context);
    }
    cleanup(context);
  }

@SuppressWarnings("unchecked")
  protected void reduce(KEYIN key, Iterable<VALUEIN> values, Context context
                        ) throws IOException, InterruptedException {
    for(VALUEIN value: values) {
      context.write((KEYOUT) key, (VALUEOUT) value);
    }