码迷，mamicode.com
首页 > 其他好文 > 详细

Map Task内部实现分析

时间：2015-12-14 01:28:51      阅读：290      评论：0      收藏：0      [点我收藏+]

标签：

转自：http://blog.csdn.net/androidlushangderen/article/details/41142795

上篇我刚刚学习完，Spilt的过程，还算比较简单的了，接下来学习的就是Map操作的过程了，Map和Reduce一样，是整个MapReduce的重要内容，所以，这一篇，我会好好的讲讲里面的内部实现过程。首先要说，MapTask，分为4种，可能这一点上有人就可能知道了，分别是Job-setup Task，Job-cleanup Task，Task-cleanup和Map Task。前面3个都是辅助性质的任务，不是本文分析的重点，我讲的就是里面的最最重要的MapTask。

        MapTask的整个过程分为5个阶段:

Read----->Map------>Collect------->Spill------>Combine

来张时序图，简单明了:

技术分享

在后面的代码分析中，你会看到各自方法的调用过程。

        在分析整个过程之前，得先了解里面的一些内部结构，MapTask类作为Map Task的一个载体，他的类关系如下:

技术分享

我们调用的就是里面的run方法，开启map任务，相应的代码:

 

[java] view plaincopyprint?
 
  1. /** 
  2.    * mapTask主要执行流程 
  3.    */  
  4.   @Override  
  5.   public void run(final JobConf job, final TaskUmbilicalProtocol umbilical)   
  6.     throws IOException, ClassNotFoundException, InterruptedException {  
  7.     this.umbilical = umbilical;  
  8.   
  9.     // start thread that will handle communication with parent  
  10.     //发送task任务报告，与父进程做交流  
  11.     TaskReporter reporter = new TaskReporter(getProgress(), umbilical,  
  12.         jvmContext);  
  13.     reporter.startCommunicationThread();  
  14.     //判断用的是新的MapReduceAPI还是旧的API  
  15.     boolean useNewApi = job.getUseNewMapper();  
  16.     initialize(job, getJobID(), reporter, useNewApi);  
  17.   
  18.     // check if it is a cleanupJobTask  
  19.     //map任务有4种，Job-setup Task, Job-cleanup Task, Task-cleanup Task和MapTask  
  20.     if (jobCleanup) {  
  21.       //这里执行的是Job-cleanup Task  
  22.       runJobCleanupTask(umbilical, reporter);  
  23.       return;  
  24.     }  
  25.     if (jobSetup) {  
  26.       //这里执行的是Job-setup Task  
  27.       runJobSetupTask(umbilical, reporter);  
  28.       return;  
  29.     }  
  30.     if (taskCleanup) {  
  31.       //这里执行的是Task-cleanup Task  
  32.       runTaskCleanupTask(umbilical, reporter);  
  33.       return;  
  34.     }  
  35.   
  36.     //如果前面3个任务都不是，执行的就是最主要的MapTask,根据新老API调用不同的方法  
  37.     if (useNewApi) {  
  38.       runNewMapper(job, splitMetaInfo, umbilical, reporter);  
  39.     } else {  
  40.       //我们关注一下老的方法实现splitMetaInfo为Spilt分片的信息，由于上步骤的InputFormat过程传入的  
  41.       runOldMapper(job, splitMetaInfo, umbilical, reporter);  
  42.     }  
  43.     done(umbilical, reporter);  
  44.   }  

在这里我研究的都是旧的API所以往runOldMapper里面跳。在这里我要插入一句，后面的执行都会围绕着一个叫Mapper的东西，就是用户执行map函数的一个代理称呼一样，他可以完全自己重写map的背后的过程，也可以用系统自带的mapp流程。

 

技术分享

系统已经给了MapRunner的具体实现:

 

[java] view plaincopyprint?
 
  1. public void run(RecordReader<K1, V1> input, OutputCollector<K2, V2> output,  
  2.                   Reporter reporter)  
  3.     throws IOException {  
  4.     try {  
  5.       // allocate key & value instances that are re-used for all entries  
  6.       K1 key = input.createKey();  
  7.       V1 value = input.createValue();  
  8.         
  9.       //从RecordReader中获取每个键值对，调用用户写的map函数  
  10.       while (input.next(key, value)) {  
  11.         // map pair to output  
  12.         //调用用户写的map函数  
  13.         mapper.map(key, value, output, reporter);  
  14.         if(incrProcCount) {  
  15.           reporter.incrCounter(SkipBadRecords.COUNTER_GROUP,   
  16.               SkipBadRecords.COUNTER_MAP_PROCESSED_RECORDS, 1);  
  17.         }  
  18.       }  
  19.     } finally {  
  20.       //结束了关闭mapper  
  21.       mapper.close();  
  22.     }  
  23.   }  

从这里我们可以看出Map的过程就是迭代式的重复的执行用户定义的Map函数操作。好了，有了这些前提，我们可以往里深入的学习了刚刚说到了runOldMapper方法，里面马上要进行的就是Map Task的第一个过程Read。

 

      Read阶段的作业就是从RecordReader中读取出一个个key-value，准备给后面的map过程执行map函数操作。

 

[java] view plaincopyprint?
 
  1. //获取输入inputSplit信息  
  2.     InputSplit inputSplit = getSplitDetails(new Path(splitIndex.getSplitLocation()),  
  3.            splitIndex.getStartOffset());  
  4.   
  5.     updateJobWithSplit(job, inputSplit);  
  6.     reporter.setInputSplit(inputSplit);  
  7.       
  8.     //是否是跳过错误记录模式,获取RecordReader  
  9.     RecordReader<INKEY,INVALUE> in = isSkipping() ?   
  10.         new SkippingRecordReader<INKEY,INVALUE>(inputSplit, umbilical, reporter) :  
  11.         new TrackedRecordReader<INKEY,INVALUE>(inputSplit, job, reporter);  

        后面的就是Map阶段，把值取出来之后，就要给Mapper去执行里面的run方法了，run方法里面会调用用户自己实现的map函数，之前也都是分析过了的。在用户编写的map的尾部，一般会调用collect.collect()方法，把处理后的key-value输出，这个时候，也就来到了collect阶段。

 

 

[java] view plaincopyprint?
 
  1. runner.run(in, new OldOutputCollector(collector, conf), reporter);  

        之后进行的是Collect阶段主要的操作时什么呢，就是把一堆堆的key-value进行分区输出到环形缓冲区中，这是的数据仅仅放在内存中，还没有写到磁盘中。在collect这个过程中涉及的东西还比较多，看一下结构关系图；

 

技术分享

里面有个partitioner的成员变量，专门用于获取key-value的的分区号，默认是通过key的哈希取模运算，得到分区号的，当然你可以自定义实现，如果不分区的话partition就是等于-1。

 

[java] view plaincopyprint?
 
  1. /** 
  2.  * Since the mapred and mapreduce Partitioners don‘t share a common interface 
  3.  * (JobConfigurable is deprecated and a subtype of mapred.Partitioner), the 
  4.  * partitioner lives in Old/NewOutputCollector. Note that, for map-only jobs, 
  5.  * the configured partitioner should not be called. It‘s common for 
  6.  * partitioners to compute a result mod numReduces, which causes a div0 error 
  7.  */  
  8. private static class OldOutputCollector<K,V> implements OutputCollector<K,V> {  
  9.   private final Partitioner<K,V> partitioner;  
  10.   private final MapOutputCollector<K,V> collector;  
  11.   private final int numPartitions;  
  12.   
  13.   @SuppressWarnings("unchecked")  
  14.   OldOutputCollector(MapOutputCollector<K,V> collector, JobConf conf) {  
  15.     numPartitions = conf.getNumReduceTasks();  
  16.     if (numPartitions > 0) {  
  17.     //如果分区数大于0,则反射获取系统配置方法，默认哈希去模，用户可以自己实现字节的分区方法  
  18.     //因为是RPC传来的，所以采用反射  
  19.       partitioner = (Partitioner<K,V>)  
  20.         ReflectionUtils.newInstance(conf.getPartitionerClass(), conf);  
  21.     } else {  
  22.     //如果分区数为0，说明不进行分区  
  23.       partitioner = new Partitioner<K,V>() {  
  24.         @Override  
  25.         public void configure(JobConf job) { }  
  26.         @Override  
  27.         public int getPartition(K key, V value, int numPartitions) {  
  28.         //分区号直接返回-1代表不分区处理  
  29.           return -1;  
  30.         }  
  31.       };  
  32.     }  
  33.     this.collector = collector;  
  34.   }  
  35.   .....  

collect的代理调用实现方法如下，注意此时还不是真正调用:

 

 

[java] view plaincopyprint?
 
  1. .....  
  2.     @Override  
  3.     public void collect(K key, V value) throws IOException {  
  4.       try {  
  5.         //具体通过collect方法分区写入内存，调用partitioner.getPartition获取分区号  
  6.         //缓冲区为环形缓冲区  
  7.         collector.collect(key, value,  
  8.                           partitioner.getPartition(key, value, numPartitions));  
  9.       } catch (InterruptedException ie) {  
  10.         Thread.currentThread().interrupt();  
  11.         throw new IOException("interrupt exception", ie);  
  12.       }  
  13.     }  

这里的collector指的是上面代码中的MapOutputCollector对象，开放给用调用的是OldOutputCollector，但是我们看看代码:

 

 

[java] view plaincopyprint?
 
  1. interface MapOutputCollector<K, V> {  
  2.   
  3.     public void collect(K key, V value, int partition  
  4.                         ) throws IOException, InterruptedException;  
  5.     public void close() throws IOException, InterruptedException;  
  6.       
  7.     public void flush() throws IOException, InterruptedException,   
  8.                                ClassNotFoundException;  
  9.           
  10.   }  


他只是一个接口，真正的实现是谁呢?这个时候应该回头看一下代码:

 

 

[java] view plaincopyprint?
 
  1. private <INKEY,INVALUE,OUTKEY,OUTVALUE>  
  2.   void runOldMapper(final JobConf job,  
  3.                     final TaskSplitIndex splitIndex,  
  4.                     final TaskUmbilicalProtocol umbilical,  
  5.                     TaskReporter reporter  
  6.                     ) throws IOException, InterruptedException,  
  7.                              ClassNotFoundException {  
  8.     ...  
  9.     int numReduceTasks = conf.getNumReduceTasks();  
  10.     LOG.info("numReduceTasks: " + numReduceTasks);  
  11.     MapOutputCollector collector = null;  
  12.     if (numReduceTasks > 0) {  
  13.       //如果存在ReduceTask，则将数据存入MapOutputBuffer环形缓冲  
  14.       collector = new MapOutputBuffer(umbilical, job, reporter);  
  15.     } else {   
  16.       //如果没有ReduceTask任务的存在，直接写入把操作结果写入HDFS作为最终结果  
  17.       collector = new DirectMapOutputCollector(umbilical, job, reporter);  
  18.     }  
  19.     MapRunnable<INKEY,INVALUE,OUTKEY,OUTVALUE> runner =  
  20.       ReflectionUtils.newInstance(job.getMapRunnerClass(), job);  
  21.   
  22.     try {  
  23.       runner.run(in, new OldOutputCollector(collector, conf), reporter);  
  24.       .....  

分为2种情况当有Reduce任务时，collector为MapOutputBuffer，没有Reduce任务时为DirectMapOutputCollector，从这里也能明白，作者考虑的很周全呢，没有Reduce直接写入HDFS，效率会高很多。也就是说，最终的collect方法就是MapOutputBuffer的方法了。

 

因为collect的操作时将数据存入环形缓冲区，这意味着，用户对数据的读写都是在同个缓冲区上的，所以为了避免出现脏数据的现象，一定会做额外处理，这里作者用了和BlockingQueue类似的操作，用一个ReetrantLocj，获取2个锁控制条件，一个为spillDone

，一个为spillReady，同个condition的await，signal方法实现丢缓冲区的读写控制。

 

[java] view plaincopyprint?
 
  1. .....  
  2.     private final ReentrantLock spillLock = new ReentrantLock();  
  3.     private final Condition spillDone = spillLock.newCondition();  
  4.     private final Condition spillReady = spillLock.newCondition();  
  5.     .....  

然后看collect的方法:

 

 

[java] view plaincopyprint?
 
  1. public synchronized void collect(K key, V value, int partition  
  2.               ) throws IOException {  
  3.       .....  
  4.       try {  
  5.         // serialize key bytes into buffer  
  6.         int keystart = bufindex;  
  7.         keySerializer.serialize(key);  
  8.         if (bufindex < keystart) {  
  9.           // wrapped the key; reset required  
  10.           bb.reset();  
  11.           keystart = 0;  
  12.         }  
  13.         // serialize value bytes into buffer  
  14.         final int valstart = bufindex;  
  15.         valSerializer.serialize(value);  
  16.         int valend = bb.markRecord();  
  17.   
  18.         if (partition < 0 || partition >= partitions) {  
  19.           throw new IOException("Illegal partition for " + key + " (" +  
  20.               partition + ")");  
  21.         }  
  22.         ....  


至于环形缓冲区的结构，不是本文的重点，结构设计还是比较复杂的，大家可以自行学习。当环形缓冲区内的数据渐渐地被填满之后，会出现"溢写"操作，就是把缓冲中的数据写到磁盘DISK中，这个过程就是后面的Spill阶段了。

 

      Spill的阶段会时不时的穿插在collect的执行过程中。

 

[java] view plaincopyprint?
 
  1. ...  
  2.           if (kvstart == kvend && kvsoftlimit) {  
  3.             LOG.info("Spilling map output: record full = " + kvsoftlimit);  
  4.             startSpill();  
  5.           }  

如果开头kvstart的位置等kvend的位置，说明转了一圈有到头了，数据已经满了的状态，开始spill溢写操作。

 

 

[java] view plaincopyprint?
 
  1. private synchronized void startSpill() {  
  2.       LOG.info("bufstart = " + bufstart + "; bufend = " + bufmark +  
  3.                "; bufvoid = " + bufvoid);  
  4.       LOG.info("kvstart = " + kvstart + "; kvend = " + kvindex +  
  5.                "; length = " + kvoffsets.length);  
  6.       kvend = kvindex;  
  7.       bufend = bufmark;  
  8.       spillReady.signal();  
  9.     }  

会触发condition的信号量操作:

 

 

[java] view plaincopyprint?
 
  1. private synchronized void startSpill() {  
  2.       LOG.info("bufstart = " + bufstart + "; bufend = " + bufmark +  
  3.                "; bufvoid = " + bufvoid);  
  4.       LOG.info("kvstart = " + kvstart + "; kvend = " + kvindex +  
  5.                "; length = " + kvoffsets.length);  
  6.       kvend = kvindex;  
  7.       bufend = bufmark;  
  8.       spillReady.signal();  
  9.     }  

就会跑到了SpillThead这个地方执行sortAndSpill方法：

 

 

[java] view plaincopyprint?
 
  1. spillThreadRunning = true;  
  2.         try {  
  3.           while (true) {  
  4.             spillDone.signal();  
  5.             while (kvstart == kvend) {  
  6.               spillReady.await();  
  7.             }  
  8.             try {  
  9.               spillLock.unlock();  
  10.               //当缓冲区溢出时，写到磁盘中  
  11.               sortAndSpill();  

sortAndSpill里面会对数据做写入文件操作写入之前还会有sort排序操作，数据多了还会进行一定的combine合并操作。

 

 

[java] view plaincopyprint?
 
  1. private void sortAndSpill() throws IOException, ClassNotFoundException,  
  2.                                        InterruptedException {  
  3.       ......  
  4.       try {  
  5.         // create spill file  
  6.         final SpillRecord spillRec = new SpillRecord(partitions);  
  7.         final Path filename =  
  8.             mapOutputFile.getSpillFileForWrite(numSpills, size);  
  9.         out = rfs.create(filename);  
  10.   
  11.         final int endPosition = (kvend > kvstart)  
  12.           ? kvend  
  13.           : kvoffsets.length + kvend;  
  14.         //在写入操作前进行排序操作  
  15.         sorter.sort(MapOutputBuffer.this, kvstart, endPosition, reporter);  
  16.         int spindex = kvstart;  
  17.         IndexRecord rec = new IndexRecord();  
  18.         InMemValBytes value = new InMemValBytes();  
  19.         for (int i = 0; i < partitions; ++i) {  
  20.           IFile.Writer<K, V> writer = null;  
  21.           try {  
  22.             long segmentStart = out.getPos();  
  23.             writer = new Writer<K, V>(job, out, keyClass, valClass, codec,  
  24.                                       spilledRecordsCounter);  
  25.             if (combinerRunner == null) {  
  26.               // spill directly  
  27.               DataInputBuffer key = new DataInputBuffer();  
  28.               while (spindex < endPosition &&  
  29.                   kvindices[kvoffsets[spindex % kvoffsets.length]  
  30.                             + PARTITION] == i) {  
  31.                 final int kvoff = kvoffsets[spindex % kvoffsets.length];  
  32.                 getVBytesForOffset(kvoff, value);  
  33.                 key.reset(kvbuffer, kvindices[kvoff + KEYSTART],  
  34.                           (kvindices[kvoff + VALSTART] -   
  35.                            kvindices[kvoff + KEYSTART]));  
  36.                 //writer中写入键值对操作  
  37.                 writer.append(key, value);  
  38.                 ++spindex;  
  39.               }  
  40.             } else {  
  41.               int spstart = spindex;  
  42.               while (spindex < endPosition &&  
  43.                   kvindices[kvoffsets[spindex % kvoffsets.length]  
  44.                             + PARTITION] == i) {  
  45.                 ++spindex;  
  46.               }  
  47.               // Note: we would like to avoid the combiner if we‘ve fewer  
  48.               // than some threshold of records for a partition  
  49.               //如果分区多的话，执行合并操作  
  50.               if (spstart != spindex) {  
  51.                 combineCollector.setWriter(writer);  
  52.                 RawKeyValueIterator kvIter =  
  53.                   new MRResultIterator(spstart, spindex);  
  54.                 //执行一次文件合并combine操作  
  55.                 combinerRunner.combine(kvIter, combineCollector);  
  56.               }  
  57.             }  
  58.   
  59.           ......  
  60.           //写入到文件中  
  61.           spillRec.writeToFile(indexFilename, job);  
  62.         } else {  
  63.           indexCacheList.add(spillRec);  
  64.           totalIndexCacheMemory +=  
  65.             spillRec.size() * MAP_OUTPUT_INDEX_RECORD_LENGTH;  
  66.         }  
  67.         LOG.info("Finished spill " + numSpills);  
  68.         ++numSpills;  
  69.       } finally {  
  70.         if (out != null) out.close();  
  71.       }  
  72.     }  

       每次Spill的过程都会产生一堆堆的文件，在最后的时候就会来到了Combine阶段，也就是Map任务的最后一个阶段了，他的任务就是把所有上一阶段的任务产生的文件进行Merge操作，合并成一个文件，便于后面的Reduce的任务的读取，在代码的对应实现中是collect.flush()方法。

 

 

[java] view plaincopyprint?
 
  1. .....  
  2.     try {  
  3.       runner.run(in, new OldOutputCollector(collector, conf), reporter);  
  4.       //将collector中的数据刷新到内存中去  
  5.       collector.flush();  
  6.     } finally {  
  7.       //close  
  8.       in.close();                               // close input  
  9.       collector.close();  
  10.     }  
  11.   }  

这里的collector的flush方法调用的就是MapOutputBuffer.flush方法，

[java] view plaincopyprint?
 
  1. public synchronized void flush() throws IOException, ClassNotFoundException,  
  2.                                             InterruptedException {  
  3.       ...  
  4.       // shut down spill thread and wait for it to exit. Since the preceding  
  5.       // ensures that it is finished with its work (and sortAndSpill did not  
  6.       // throw), we elect to use an interrupt instead of setting a flag.  
  7.       // Spilling simultaneously from this thread while the spill thread  
  8.       // finishes its work might be both a useful way to extend this and also  
  9.       // sufficient motivation for the latter approach.  
  10.       try {  
  11.         spillThread.interrupt();  
  12.         spillThread.join();  
  13.       } catch (InterruptedException e) {  
  14.         throw (IOException)new IOException("Spill failed"  
  15.             ).initCause(e);  
  16.       }  
  17.       // release sort buffer before the merge  
  18.       kvbuffer = null;  
  19.       //最后进行merge合并成一个文件  
  20.       mergeParts();  
  21.       Path outputPath = mapOutputFile.getOutputFile();  
  22.       fileOutputByteCounter.increment(rfs.getFileStatus(outputPath).getLen());  
  23.     }  

至此，Map任务宣告结束了，整体流程还是真是有点九曲十八弯的感觉。分析这么一个比较庞杂的过程，我一直在想如何更好的表达出我的想法，欢迎MapReduce的学习者，提出意见，共同学习

Map Task内部实现分析

标签：

原文地址：http://www.cnblogs.com/cxzdy/p/5043994.html
踩
(0)
赞
(0)
   
举报
评论 一句话评论（0）
登录后才能评论！
分享档案
更多>
2021年07月29日 (22)
2021年07月28日 (40)
2021年07月27日 (32)
2021年07月26日 (79)
2021年07月23日 (29)
2021年07月22日 (30)
2021年07月21日 (42)
2021年07月20日 (16)
2021年07月19日 (90)
2021年07月16日 (35)
周排行
mamicode.com排行更多图片
更多
友情链接
兰亭集智   国之画   百度统计   站长统计   阿里云   chrome插件   新版天听网
关于我们 - 联系我们 - 留言反馈
© 2014 mamicode.com 版权所有  联系我们:gaon5@hotmail.com
迷上了代码！