首页 诗词 字典 板报 句子 名言 友答 励志 学校 网站地图
当前位置: 首页 > 教程频道 > 服务器 > 云计算 >

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

2013-03-26 
hadoop核心逻辑shuffle代码分析-map端首先要推荐一下:http://www.alidata.org/archives/1470阿里的大牛在

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

首先要推荐一下:http://www.alidata.org/archives/1470

阿里的大牛在上面的文章中比较详细的介绍了shuffle过程中mapper和reduce的每个过程,强烈推荐先读一下。

 

不过,上文没有写明一些实现的细节,比如:spill的过程,mapper生成文件的partition是怎么做的等等,相信有很多人跟我一样在看了上面的文章后还是有很多疑问,我也是带着疑问花了很久的看了cdh4.1.0版本shuffle的逻辑,整理成本文,为以后回顾所用。

 

 首先用一张图展示下map的流程:

 hadoop核心逻辑shuffle代码分析-地图端在上图中,我们假设此次mapreduce有多个mapper和2个reducer,p0 p1分别代表该数据应该分配到哪个reducer端。我将mapper的过程大致分为5个过程。 1.prepare Input。Mapreduce程序都需要指定输入文件,输入的格式有很多种,最常见的是保存在hdfs上的文本文件。在用户提交job到jobtrack(ResourceManager)前的job就会根据用户的输入文件计算出需要多少mapper,多少reducer,mapper的输入InputSplit有多大,block块名称等。mapper在prepare input阶段只需要根据inputFormat类型创建对应的RecordReader打开对应的inputSplit分片即可。如果job配置了combiner还需初始化combiner。代码见MapTask类run方法 2.mapper process这里的mapper指用户使用或自己继承的mapper类,这也是所有初学mapreduce的同学首先看到的类。
  /**   * Called once for each key/value pair in the input split. Most applications   * should override this, but the default is the identity function.   */  @SuppressWarnings("unchecked")  protected void map(KEYIN key, VALUEIN value,                      Context context) throws IOException, InterruptedException {    context.write((KEYOUT) key, (VALUEOUT) value);  }
可以看到mapper默认的map方法就是取出key,value并放到context对象中。context对象包装了一个内存中的buf,下面会介绍。
public void run(Context context) throws IOException, InterruptedException {    setup(context);    while (context.nextKeyValue()) {      map(context.getCurrentKey(), context.getCurrentValue(), context);    }    cleanup(context);  }
run方法就是mapper实际运行的过程:不停的从context的inputSplit对象中取出keyvalue对,通过map方法处理再保存到context包装的内存buf中。 3.buffer in memery
key value在写入context中后实际是写入MapOutputBuffer类中。在第一个阶段的初始化过程中,MapOutputBuffer类会根据配置文件初始化内存buffer,我们来看下都有哪些参数:
partitions = job.getNumReduceTasks();      rfs = ((LocalFileSystem)FileSystem.getLocal(job)).getRaw();      //sanity checks      final float spillper =        job.getFloat(JobContext.MAP_SORT_SPILL_PERCENT, (float)0.8);      final int sortmb = job.getInt(JobContext.IO_SORT_MB, 100);      indexCacheMemoryLimit = job.getInt(JobContext.INDEX_CACHE_MEMORY_LIMIT,                                         INDEX_CACHE_MEMORY_LIMIT_DEFAULT);      if (spillper > (float)1.0 || spillper <= (float)0.0) {        throw new IOException("Invalid \"" + JobContext.MAP_SORT_SPILL_PERCENT +            "\": " + spillper);      }      if ((sortmb & 0x7FF) != sortmb) {        throw new IOException(            "Invalid \"" + JobContext.IO_SORT_MB + "\": " + sortmb);      }      sorter = ReflectionUtils.newInstance(job.getClass("map.sort.class",            QuickSort.class, IndexedSorter.class), job);
partition:mapper的数据需要分配到reduce端的个数,由用户的job指定,默认为1.spillper:内存buf使用到此比例就会触发spill,将内存中的数据flush成一个文件。默认为0.8sortmb:内存buf的大小,默认100MBindexCacheMemoryLimit:内存index的大小。默认为1024*1024sorter:对mapper输出的key的排序,默认是快排 内存buffer比较复杂,贴一张图介绍一下这块内存buf的结构:hadoop核心逻辑shuffle代码分析-地图端当一对keyvalue写入时首先会从wrap buf的右侧开始往左写,同时,会把一条keyvalue的meta信息(partition,keystart,valuestart)写入到最左边的index区域。当wrap buf大小达到spill的触发比例后会block写入,挖出一部分数据开始spill,直到spill完成后才能继续写,不过写入位置不会置零,而是类似循环buf那样,在spill掉数据后可以重复利用内存中的buf区域。 这里单独讲一下partition:
@Override    public void write(K key, V value) throws IOException, InterruptedException {      collector.collect(key, value,                        partitioner.getPartition(key, value, partitions));    }

在keyvalue对写入MapOutputBuffer时会调用partitioner.getPartition方法计算partition即应该分配到哪个reducer,这里的partition只是在内存的buf的index区写入一条记录而已,和下一个部分的partition不一样哦。看下默认的partitioner:HashPartition

/** Use {@link Object#hashCode()} to partition. */  public int getPartition(K key, V value,                          int numReduceTasks) {    return (key.hashCode() & Integer.MAX_VALUE) % numReduceTasks;  }

HashPartition只是把key hash后按reduceTask的个数取模,因此一般来说,不同的key分配到哪个reducer是随即的!所以,reducer内的所有数据是有序的,但reducer之间的数据却是乱序的!要想数据整体排序,要不只设一个reducer,要不使用TotalOrderPartitioner!

 4.Partition Sort Store在第四步中,partition是和sort一起做的,负责Spill的线程在拿到一段内存buf后会调用QuickSort的sort方法进行内存中的快排。
sorter.sort(MapOutputBuffer.this, mstart, mend, reporter);
排序的算法是先按keyvalue记录的partition排序后按key的compare方法:
public int compare(final int mi, final int mj) {      final int kvi = offsetFor(mi % maxRec);      final int kvj = offsetFor(mj % maxRec);      final int kvip = kvmeta.get(kvi + PARTITION);      final int kvjp = kvmeta.get(kvj + PARTITION);      // sort by partition      if (kvip != kvjp) {        return kvip - kvjp;      }      // sort by key      return comparator.compare(kvbuffer,          kvmeta.get(kvi + KEYSTART),          kvmeta.get(kvi + VALSTART) - kvmeta.get(kvi + KEYSTART),          kvbuffer,          kvmeta.get(kvj + KEYSTART),          kvmeta.get(kvj + VALSTART) - kvmeta.get(kvj + KEYSTART));    }
因此,mapper输出的keyvalue首先是按partition聚合。而我们如果指定key的compare方法会在这里生效并进行排序。最后,一次spill的输出文件类似下图。hadoop核心逻辑shuffle代码分析-地图端在对内存中的buf排序后开始写文件。
for (int i = 0; i < partitions; ++i) {          IFile.Writer<K, V> writer = null;          try {            long segmentStart = out.getPos();            writer = new Writer<K, V>(job, out, keyClass, valClass, codec,                                      spilledRecordsCounter);            if (combinerRunner == null) {              // spill directly              DataInputBuffer key = new DataInputBuffer();              while (spindex < mend &&                  kvmeta.get(offsetFor(spindex % maxRec) + PARTITION) == i) {                final int kvoff = offsetFor(spindex % maxRec);                key.reset(kvbuffer, kvmeta.get(kvoff + KEYSTART),                          (kvmeta.get(kvoff + VALSTART) -                           kvmeta.get(kvoff + KEYSTART)));                getVBytesForOffset(kvoff, value);                writer.append(key, value);                ++spindex;              }            } else {              int spstart = spindex;              while (spindex < mend &&                  kvmeta.get(offsetFor(spindex % maxRec)                            + PARTITION) == i) {                ++spindex;              }              // Note: we would like to avoid the combiner if we've fewer              // than some threshold of records for a partition              if (spstart != spindex) {                combineCollector.setWriter(writer);                RawKeyValueIterator kvIter =                  new MRResultIterator(spstart, spindex);                combinerRunner.combine(kvIter, combineCollector);              }            }
如果job没有定义combiner则直接写文件,如果有combiner则在这里进行combine。
在生成spill文件后还会将此次spillRecord的记录写在一个index文件中。
Path indexFilename =              mapOutputFile.getSpillIndexFileForWrite(numSpills, partitions                  * MAP_OUTPUT_INDEX_RECORD_LENGTH);          spillRec.writeToFile(indexFilename, job);
rec.startOffset = segmentStart;            rec.rawLength = writer.getRawLength();            rec.partLength = writer.getCompressedLength();            spillRec.putIndex(rec, i);
 5.merge当mapper执行完毕后,就进入merge阶段。首先看下相关的配置参数:
int mergeFactor = job.getInt(JobContext.IO_SORT_FACTOR, 100);
mergeFactor:同时merge的文件数。 merge阶段的目的是将多个spill生成的中间文件合并为一个输出文件,这里的合并不同于combiner,无论有没有配置combiner这里的merge都会执行。merge阶段的输出是一个数据文件MapFinalOutputFile和一个index文件。看下相关代码:
RawKeyValueIterator kvIter = Merger.merge(job, rfs,                         keyClass, valClass, codec,                         segmentList, mergeFactor,                         new Path(mapId.toString()),                         job.getOutputKeyComparator(), reporter, sortSegments,                         null, spilledRecordsCounter, sortPhase.phase());          //write merged output to disk          long segmentStart = finalOut.getPos();          Writer<K, V> writer =              new Writer<K, V>(job, finalOut, keyClass, valClass, codec,                               spilledRecordsCounter);          if (combinerRunner == null || numSpills < minSpillsForCombine) {            Merger.writeFile(kvIter, writer, reporter, job);          } else {            combineCollector.setWriter(writer);            combinerRunner.combine(kvIter, combineCollector);          }
说下merge的算法。每个spill生成的文件中keyvalue都是有序的,但不同的文件却是乱序的,类似多个有序文件的多路归并算法。Merger分别取出需要merge的spillfile的最小的keyvalue,放入一个内存堆中,每次从堆中取出一个最小的值,并把此值保存到merge的输出文件中。这里和hbase中scan的算法非常相似,在分布式系统中多路归并排序真是当红小生啊!这里merge时不同的partition的key是不会比较的,只有相同的partition的keyvalue才会进行排序和合并。最后的输出文件类似下图。hadoop核心逻辑shuffle代码分析-地图端如果用户定义了combiner,在merge的过程中也会进行combine,因为虽然第四步中combine过但那只是部分输入的combine,在merge时仍然需要combine。这里有人问了,既然这里有combiner,为啥在spill输出时还要combine纳,我认为是因为每次combine都会大大减少输出文件的大小,spill时就combine能减少一定的IO操作。 在merge完后会把不同partition的信息保存进一个index文件以便之后reducer来拉自己部分的数据。
// record offsets          rec.startOffset = segmentStart;          rec.rawLength = writer.getRawLength();          rec.partLength = writer.getCompressedLength();          spillRec.putIndex(rec, parts);

最后,我们再对mapper过程中的要点总结一下:1.对map输出<key,value>的分区(partition)是在写入内存buf前就做好的了,方法是对key的hash。我们可以通过继承Partitioner类自己实现分区,将自己想要的数据分到同一个reducer中。2.写入内存buf速度是非常快的,但spill过程会block写入。因此,对内存buf相关参数的调优是mapreduce调优的重点之一。3.对数据的排序是基于MapOutKey排序的,因此,我们可以重载对应的方法实现customize的排序顺序4.combine在spill和merge中都是进行。多次的combine会减少mapreduce中的IO操作,如果使用得当会很好的提高性能。但需要注意的是要深刻理解combine的意义,比如平均值就不适合用combine。

热点排行