hadoop核心逻辑shuffle代码分析-reduce端

uf123

　　接上篇文章，下面是reduce端的过程分析。
　　大概介绍下reduce的实际作用。以mapreduce经常做的groupby为例，map是将输入按group by的key排序，reduce就是做各种类型的聚合，比如sum，max，mean等。因此，可想而知，reduce的输入必须是按照groupby排序的，所以自然，reduce的输入必须汇聚所有map的输入，这也是reduce框架最复杂的部分。
　　首先，还是看一张reduce过程整体时序图：
　　1.从多个mapper端fetch数据。（copy）
　　fetch指从mapper端拉取mapper的输出数据。在fetch阶段reducetask会启动多个fetch线程从所有的mapper端取这个reducer的数据，同时还有一个fetchEvent线程负责获取mapper完成的event通知Fetch线程。
　　这里顺便提下，mapreduce的调度系统有一个假设：数据就近原则，也就是yarn的控制器会根据数据的位置分配mapper的位置，因此一般来说mapper的输入数据都是本地文件，再加上配置的机架感知和本地读策略mapper的输入处理效率是很高的，但reduce就不一样了，每个reducer的输入都要来自所有的mapper，因此，一般没办法优化reducer的位置。
　　reducer不会等到所有的mapper执行完毕再去拉数据，而是在mappertask完成一定比例后就会开始fetch，下面先列出一些fetch阶段的重要参数：
　　mapreduce.job.reduce.slowstart.completedmaps：在maptask完成了一定百分比后将触发fetch，默认为0.05
　　mapreduce.reduce.shuffle.read.timeout：fetch读数据的timeout时间，默认为3分钟
　　mapreduce.reduce.shuffle.maxfetchfailures：fetch最大的失败次数，默认为10次
　　mapreduce.reduce.shuffle.connect.timeout：fetch建立连接的timeout时间，默认也是3分钟
　　mapreduce.reduce.shuffle.parallelcopies：同时创建的fetch线程个数
　　fetch的目的有2个：内存buf和磁盘。那么如何判刑是保存进内存还是硬盘纳：首先判断内存buf大小有没有超过maxSingleShuffleLimit，如果没有则存入内存，如果整个内存buf都存满了则copy过程需要暂停一会儿（stallShuffle），其他情况则写入磁盘。代码如下：
　　[java] view plaincopy

• 　　if (!canShuffleToMemory(requestedSize)) {
  
• 　　LOG.info(mapId + ": Shuffling to disk since " + requestedSize +
  
• 　　" is greater than maxSingleShuffleLimit (" +
  
• 　　maxSingleShuffleLimit + ")");
  
• 　　return new MapOutput(mapId, this, requestedSize, jobConf,
  
• 　　localDirAllocator, fetcher, true,
  
• 　　mapOutputFile);
  
• 　　}
  
• 
  
• 　　if (usedMemory > memoryLimit) {
  
• 　　LOG.debug(mapId + ": Stalling shuffle since usedMemory (" + usedMemory
  
• 　　+ ") is greater than memoryLimit (" + memoryLimit + ")." +
  
• 　　" CommitMemory is (" + commitMemory + ")");
  
• 　　return stallShuffle;
  
• 　　}
  
• 
  
• 　　// Allow the in-memory shuffle to progress
  
• 　　LOG.debug(mapId + ": Proceeding with shuffle since usedMemory ("
  
• 　　+ usedMemory + ") is lesser than memoryLimit (" + memoryLimit + ")."
  
• 　　+ "CommitMemory is (" + commitMemory + ")");
  
• 　　return unconditionalReserve(mapId, requestedSize, true);
  

　　2.内存merge和磁盘merge（merge & sort）
　　第一阶段和第二阶段其实是密切相关的。fetch线程负责取数据，merge负责存数据，merge的输出就是用户的reducer的输入。分布式环境中每个mapper的输入只是整体的一小部分，因此关键的数据的汇总就是在merge阶段处理的。reducer的merge是最复杂的，包括内存merge，磁盘merge，finalmerge等阶段一步步的将零散的mapper的输出整合成一个整体有序的输入，下面会仔细介绍此过程。
　　在第一部中提到了内存buf和磁盘，我们知道写内存的速度肯定是比磁盘快的多的，但内存也不能太大，所以需要对内存的使用做一些限制。下面列一下merge过程中的相关参数：
　　mapreduce.reduce.shuffle.input.buffer.percent：copy阶段内存使用的最大值，默认为0.9
　　mapreduce.task.io.sort.factor：每次merge的文件个数，关系到需要merge次数的参数，默认为100

　　mapreduce.reduce.shuffle.memory.limit.percent：每个fetch取到的输出的大小能够占的内存比的大小。默认是0.25。因此实际每个fetcher的输出能放在内存的大小是reducer的java heap>　　mapreduce.reduce.merge.memtomem.enabled：是否enable内存merge，默认是false，如果设置为true，则在finalmerge时内存的数据不会存入磁盘再merge而是直接在内存中merge。不过，好像内存中的merge并没有想象的那么美好，因此默认是关闭的：https://issues.apache.org/jira/browse/MAPREDUCE-268

　　mapreduce.reduce.shuffle.merge.percent：触发merge的内存条件，默认是java heap>　　下面看下InMemoryMerge处理的过程。InMemoryMerger不是上面提到的MemToMemMerge，InMemoryMerger主要负责将mapper的输出（内存buf）整合并输出到磁盘上。输出磁盘的过程类似mapper端的spill，而且也是等内存攒到一定数量后才会触发spill，但是不会出现类似mapper端的情况，因为如果超过内存的上线后mapper的输出会直接输出到磁盘上。InMemoryMerger的spill会在达到merge的threshold后触发spill。看下spill的代码：
　　[java] view plaincopy

• 　　Writer writer =
  
• 　　new Writer(jobConf, rfs, outputPath,
  
• 　　(Class) jobConf.getMapOutputKeyClass(),
  
• 　　(Class) jobConf.getMapOutputValueClass(),
  
• 　　codec, null);
  
• 
  
• 　　RawKeyValueIterator rIter = null;
  
• 　　try {
  
• 　　LOG.info("Initiating in-memory merge with " + noInMemorySegments +
  
• 　　" segments...");
  
• 
  
• 　　rIter = Merger.merge(jobConf, rfs,
  
• 　　(Class)jobConf.getMapOutputKeyClass(),
  
• 　　(Class)jobConf.getMapOutputValueClass(),
  
• 　　inMemorySegments, inMemorySegments.size(),
  
• 　　new Path(reduceId.toString()),
  
• 　　(RawComparator)jobConf.getOutputKeyComparator(),
  
• 　　reporter, spilledRecordsCounter, null, null);
  
• 
  
• 　　if (null == combinerClass) {
  
• 　　Merger.writeFile(rIter, writer, reporter, jobConf);
  
• 　　} else {
  
• 　　combineCollector.setWriter(writer);
  
• 　　combineAndSpill(rIter, reduceCombineInputCounter);
  
• 　　}
  

　　有2点需要注意：1，merge的Comparator是key的Comparator，相关原因在后面会讲到。2，merge过程也会使用combiner，原来reduce端也会combine！这个在所有的mapreduce教材中都没有讲到。
　　最后将内存buf中的值merge到一个输出文件中。DiskMerge的merge过程类似，不过开始merge的条件是磁盘上的零散文件超过mapreduce.task.io.sort.factor的2倍-1。
　　[java] view plaincopy
　　

• 　　synchronized (onDiskMerger) {
  
• 　　if (!onDiskMerger.isInProgress() &&
  
• 　　onDiskMapOutputs.size() >= (2 * ioSortFactor - 1)) {
  
• 　　onDiskMerger.startMerge(onDiskMapOutputs);
  
• 　　}
  
• 　　}
  

　　在fetch结束后会进入finalmerge。所谓finalmerge就是将各个mapper的输出整合成真正reducer的输入。在fetch阶段我们已经一边copy数据一边merge，因此到了finalmerge时，mapper的数据部分存在于内存buf中部分存在于tmp目录的文件中。下面先看下finalmerge的代码：
　　[java] view plaincopy

• 　　if (numMemDiskSegments > 0 &&
  
• 　　ioSortFactor > onDiskMapOutputs.size()) {
  
• 
  
• 　　mergePhaseFinished = true;
  
• 　　// must spill to disk, but can't retain in-mem for intermediate merge
  
• 　　final Path outputPath =
  
• 　　mapOutputFile.getInputFileForWrite(mapId,
  
• 　　inMemToDiskBytes).suffix(
  
• 　　Task.MERGED_OUTPUT_PREFIX);
  
• 　　final RawKeyValueIterator rIter = Merger.merge(job, fs,
  
• 　　keyClass, valueClass, memDiskSegments, numMemDiskSegments,
  
• 　　tmpDir, comparator, reporter, spilledRecordsCounter, null,
  
• 　　mergePhase);
  
• 　　final Writer writer = new Writer(job, fs, outputPath,
  
• 　　keyClass, valueClass, codec, null);
  
• 　　try {
  
• 　　Merger.writeFile(rIter, writer, reporter, job);
  
• 　　// add to list of final disk outputs.
  
• 　　onDiskMapOutputs.add(outputPath);
  

　　这段代码处理了内存中的mapper数据，将内存的数据排序和merge（其实merge可以单纯看做排序）。
　　[java] view plaincopy

• 　　// segments on disk
  
• 　　List diskSegments = new ArrayList();
  
• 　　long onDiskBytes = inMemToDiskBytes;
  
• 　　Path[] onDisk = onDiskMapOutputs.toArray(new Path[onDiskMapOutputs.size()]);
  
• 　　for (Path file : onDisk) {
  
• 　　onDiskBytes += fs.getFileStatus(file).getLen();
  
• 　　LOG.debug("Disk file: " + file + " Length is " +
  
• 　　fs.getFileStatus(file).getLen());
  
• 　　diskSegments.add(new Segment(job, fs, file, codec, keepInputs,
  
• 　　(file.toString().endsWith(
  
• 　　Task.MERGED_OUTPUT_PREFIX) ?
  
• 　　null : mergedMapOutputsCounter)
  
• 　　));
  
• 　　}
  
• 　　LOG.info("Merging " + onDisk.length + " files, " +
  
• 　　onDiskBytes + " bytes from disk");
  
• 　　Collections.sort(diskSegments, new Comparator() {
  
• 　　public int compare(Segment o1, Segment o2) {
  
• 　　if (o1.getLength() == o2.getLength()) {
  
• 　　return 0;
  
• 　　}
  
• 　　return o1.getLength() < o2.getLength() ? -1 : 1;
  
• 　　}
  
• 　　});
  

　　将磁盘上的文件信息读入内存进行最终merge前的准备。
　　[java] view plaincopy
　　

• 　　// build final list of segments from merged backed by disk + in-mem
  
• 　　List finalSegments = new ArrayList();
  
• 　　long inMemBytes = createInMemorySegments(inMemoryMapOutputs,
  
• 　　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();
  
• 　　// Pass mergePhase only if there is a going to be intermediate
  
• 　　// merges. See comment where mergePhaseFinished is being set
  
• 　　Progress thisPhase = (mergePhaseFinished) ? null : mergePhase;
  
• 　　RawKeyValueIterator diskMerge = Merger.merge(
  
• 　　job, fs, keyClass, valueClass, diskSegments,
  
• 　　ioSortFactor, numInMemSegments, tmpDir, comparator,
  
• 　　reporter, false, spilledRecordsCounter, null, thisPhase);
  
• 　　diskSegments.clear();
  
• 　　if (0 == finalSegments.size()) {
  
• 　　return diskMerge;
  
• 　　}
  
• 　　finalSegments.add(new Segment(
  
• 　　new RawKVIteratorReader(diskMerge, onDiskBytes), true));
  
• 　　}
  
• 　　return Merger.merge(job, fs, keyClass, valueClass,
  
• 　　finalSegments, finalSegments.size(), tmpDir,
  
• 　　comparator, reporter, spilledRecordsCounter, null,
  
• 　　null);
  

　　最后开始build最终merge的list。可以看到是内存buf和磁盘文件的整合。这里Merger.merge方法比较复杂，大概介绍下：对于reducer的merge不会真的全做merge，而是根据输入segments的个数做判断，如果segment个数超过io.factor（就是在mapper端详解中介绍的同时merge的个数），Merger会构造内存heap用多路归并算法合并输入的segment，直到segment的数量不超过io.factor为止，注意，这里不需要merge为一个文件。比如：输入为45个文件，io.factor为10，第一次merge前10个文件，第二次merge10到20文件，merge4次后剩余7个小于10，这时，此次merge就结束了。这样可以省一次merge，而且这次merge的IO操作还是最大的，大大的提高了效率！这也是为什么有时候要适当调高io.factor参数的原因了。
　　这里还有一个疑问的，有的文章说可以让merge的输出也就是reducer的输入直接放在内存中，我实在没找到在哪儿，因为在第一步merge内存segment时会write成磁盘文件。因此内存segment始终是要落到磁盘才行的。
　　最后，有的人可能要问了，merge结果不是一个文件那读的时候怎么办，跟merge过程类似，读的时候会把merge结果的所有文件的数据放到内存queue中（单个文件已经有序），依次取出。
　　3.执行reduce过程（groupby and exec）
　　到了reducer的执行其实就比较明了了，就是从merge的结果构成的heap中取数据，一条记录是key和相同key对应的序列。
　　[java] view plaincopy

• 　　public void run(Context context) throws IOException, InterruptedException {
  
• 　　setup(context);
  
• 　　while (context.nextKey()) {
  
• 　　reduce(context.getCurrentKey(), context.getValues(), context);
  
• 　　// If a back up store is used, reset it
  
• 　　Iterator iter = context.getValues().iterator();
  
• 　　if(iter instanceof ReduceContext.ValueIterator) {
  
• 　　((ReduceContext.ValueIterator)iter).resetBackupStore();
  
• 　　}
  
• 　　}
  
• 　　cleanup(context);
  
• 　　}
  

　　这边有个要注意的是，在0.20后的mapreduce多了一个比较的comparator：getOutputValueGroupingComparator()，在用户定义了此comparator后，reducer是用它来做grouping的。相关内容可以参考secendarySort：
　　http://www.cnblogs.com/xuxm2007/archive/2011/09/03/2165805.html
　　最后还是总结一下：
　　1.mapreduce的优化通常围绕shuffle的过程展开，包括如何增加并发、多用内存少用磁盘、减少shuffle的文件大小等，相关参数参考上文。
　　2.根据经验，shuffle时如果数据过大经常会发生OutOfMemory，这时可以适当调小相应的参数，比如减少每个fetch占内存的百分比让fetch的输出直接进入磁盘。
　　3.io.factor在mapper很多时比较影响系统的效率，不但mapper端的merge，reducer的merge也会受到影响，需要综合考虑。