Hadoop 的Combiner调用时间点

  Combiner是在Map端被执行，共有两个时机会被触发：

         ① 从环形缓冲器溢写分区文件的时候

         ② 合并溢写分区文件的时候

 

1. 初始化combinerRunner和combineCollector

MapTask.run()

  ┟ runNewMapper(job, split, umbilical, reporter);

     ┟ output = new NewOutputCollector(taskContext, job, umbilical, reporter);

Java代码  

1. if(job.getNumReduceTasks() == 0) {  
2.   output = new NewDirectOutputCollector(taskContext, job, umbilical, reporter);  
3. } else {  
4.   // 如果有reduce task，才会有Combiner task  
5.   output = new NewOutputCollector(taskContext, job, umbilical, reporter);  
6. }  

       ┟ collector = new MapOutputBuffer<K,V>(umbilical, job, reporter);

           ┟ 初始化combinerRunner和combineCollector

Java代码  

1. combinerRunner = CombinerRunner.create(job, getTaskID(),   
2.                      combineInputCounter,  
3.                      reporter, null);  
4. if (combinerRunner != null) {  
5.   combineCollector= new CombineOutputCollector<K,V>(combineOutputCounter);  
6. } else {  
7.   combineCollector = null;  
8. }  

2. Combiner调用点1：磁盘溢写（spill）

磁盘溢写会触发Combiner，有两个地方会触发溢写操作：

1. 输出Key-value到缓冲器
2. 关闭map函数输出流，执行flush方法时

2.1 输出Key-Value到缓冲器

MapTask.run()

  ┟ runNewMapper(job, split, umbilical, reporter);

    ┟ mapper.run(mapperContext);

      ┟ map(context.getCurrentKey(), context.getCurrentValue(), context); // map函数

        ┟ context.write((KEYOUT) key, (VALUEOUT) value); // map函数输出值
          ┟ NewOutputCollector.write()

            ┟ MapOutputBuffer.collect()

              ┟ startSpill();

Java代码  

1. if (kvstart == kvend && kvsoftlimit) {  
2.   LOG.info("Spilling map output: record full = " + kvsoftlimit);  
3.   startSpill(); //  缓冲器达到spill条件，溢写到磁盘  
4. }  

               ┟ spillReady.signal(); // 唤起spill线程

SpillThread.run()

  ┟ sortAndSpill();

Java代码  

1. public void run() {  
2. try {  
3. ...  
4.   while (kvstart == kvend) {  
5.     spillReady.await(); // 等待被唤醒  
6.   }  
7. ...  
8. sortAndSpill();   
9. ...  

     ┟ combinerRunner.combine(kvIter, combineCollector); // 运行combiner

Java代码  

1. int spstart = spindex; // spstart为kvoffet数组start, spindex为kvoffset数组end  
2. while (spindex < endPosition &&  
3.   kvindices[kvoffsets[spindex % kvoffsets.length]  
4.         + PARTITION] == i) {  
5. ++spindex;  
6. }  
7. // Note: we would like to avoid the combiner if we've fewer  
8. // than some threshold of records for a partition  
9. // 如果start == end，说明该分区只有一条记录，则不进行combiner操作；否则执行combiner  
10. if (spstart != spindex) {  
11.   combineCollector.setWriter(writer);  
12.   RawKeyValueIterator kvIter = new MRResultIterator(spstart, spindex);  
13.   combinerRunner.combine(kvIter, combineCollector);  
14. }  

2.2 map输出流flush方法

MapTask.run()

  ┟ runNewMapper(job, split, umbilical, reporter);

    ┟ output.close(mapperContext); // 关闭map输出流

      ┟ NewOutputCollector.close();

        ┟ collector.flush();

          ┟ MapOutputBuffer.flush()

            ┟ sortAndSpill(); 运行combiner，同上

3. Combiner调用点2：map端分区文件合并

MapTask.run()

  ┟ runNewMapper(job, split, umbilical, reporter);

    ┟ output.close(mapperContext); // 关闭map输出流

      ┟ NewOutputCollector.close();

        ┟ collector.flush();

          ┟ MapOutputBuffer.flush()

              ┟ mergeParts();

Java代码  

1. // minSpillsForCombine 在MapOutputBuffer构造函数内被初始化，  
2. // numSpills 为mapTask已经溢写到磁盘spill文件数量  
3. if (combinerRunner == null || numSpills < minSpillsForCombine) {  
4.   Merger.writeFile(kvIter, writer, reporter, job);  
5. } else {  
6.   combineCollector.setWriter(writer);  
7.   combinerRunner.combine(kvIter, combineCollector); <- 执行combiner  
8. }  

 --end