Map/Reduce执行流程

布满荆棘的人生 2022-04-21 18:58 224阅读 0赞

\[b\]\[color=green\]\[size=large\]FileSplit:文件的子集--文件分割体  
简介：  
  
这篇文档描述在hadoop中map和reduce操作是怎样具体完成的。如果你对Google的MapReduce各式模式不熟悉，请先参阅MapReduce--http://labs.google.com/papers/mapreduce.html  
  
Map  
  
由于Map是并行地对输入的文件集进行操作，所以它的第一步(FileSplit) 就是把文件集分割成一些子集.如果一个单个的文件大到它已影响到查找效率时，它会被分割成一些小的分割体。要指出的是分割这个一步是不知道输入文件的内部逻辑结构的，比如，以行为逻辑分割的文本文件会被以任意的字节界限分割，所以这个具体分割要自己去指定也可以用hadoop已经定义的几个简单分割。然后每个文件分割体都会对应地有一个新的map任务。  
  
当单个map任务开始时，它会对每个配置过的reduce任务开启一个新的输出书写器(writer).紧接着它(writer)会用从指定的特定InputFormat里得到的RecordReader去读它的文件分割体。InputFormat类分析输入文件并产生key-value键值对.同时InputFormat必需要处理在以文件分割时边界处的记录。比如TextInputFormat会读取分割边界的文件分割体有最后一行，如果当读取的分割体不是第一个时，TextInputFormat会忽略第一行的内容。  
  
  
InputFormat类并不需要产生一些对有意义的键值对。比如TextInputFormat类的默认输出是以输入文本的行内容为value,以行偏移量为key--大多数应用只用到而很少用到偏移量。  
  
传给用户配置的mapper的键值对都是从RecordReader读入的，用户提供的Mapper类就可以对键值对进行任意的操作再调用OutputCollector.collect方法来重新收集自己定义后的键值对。产生的输出必需用一个Key类和一个value类，这是因为Map的输出结果要被以SequenceFile的形式写入磁盘，这种形式包括每个文件的类型信息和所有的记录都是同类形的(如果你想输出不同的数据结构你可以继承个子类出来)。Map的输入和输出键值对不需要在类型上有联系.  
  
当Mapper的输出被收集后，它们会被Partitioner类以指定的方式区分地写出到输出文件里。默认是以HashPartitioner类用key类的哈希函数产生的hashcode来区分(因此就要有一个很好的哈希函数，才可以使在各个reduce任务时负载匀衡)。详细可以查看MapTask类。N个输入可以产生M个map任务去跑，每个map任务会产生配置的reduce任务数个输出文件。每个输出文件都会面向一个特定的reduce任务同时所有从map任务产生的键值对都会被送到reduce里。所以在一个特定的reduce任务中对于一个给定的key所有的键值对都会被处理。  
  
  
  
  
Combine  
  
当map操作输出它的键值对时他们已经在内存中存在了。为了性能和效率的考虑，有时候提供一个拥有reduce功能的合成器是有好处的。如果有合成器，那么map的键值对就不会被马上写入到输出里，他们会被收集在list里，一个key值一个list，当写入一定数量的键值对时，这部分缓冲会被送进合成器，每个key都的所有value都会被送进合成器的reduce方法里并且就像原先map输出的键值对一样。  
  
  
  
  
比如，hadoop案例中的wordcount程序，它的map操作输出是(word,1)键值对，在输入中的词的计数可以用合成器来加速这个操作。一个合成操作会在内存中收集处理lists，一个词一个list。当一定数量的键值对输出到内存中时，就调用合成操作的reduce方法，每次都以一个唯一的词为key，values是list的迭代器。然后合成器输出(word,count-in-this-part-of-the-input)键值对。从Reduce操作的观点来说合成器也拥有Map输出中相同的信息，但是这样会比原先远远减少硬盘的读写。  
  
Reduce  
  
当一个reduce任务开始时，它的输入是分散在各个节点上的map的输出文件里。如果在分布式的模式下，他们需要先在拷贝步骤里拷贝到本地文件系统上。详细可以查看ReduceTaskRunner类  
  
一旦所有的数据都在本地有效时，它会在添加步骤里加到一个文件里。然后这个文件会被合并分类这样相同的key的键值对就可以排在一起(分类步骤)。这样可以使真正的reduce操作变得简单，这个文件会被顺序地读入，值(values)会从输入文件里用一个迭代器传给reduce方法-直到下一个key。详细可以查看ReduceTask类。  
  
最后，输出由每个reduce任务的输出文件组成。面他们的格式可以由JobConf.setOutputFormat类指定，如果用到JobConf.setOutputFormat类，那么输出的key类和value类都要同时指定。  
  
  
  
我们以wordcount为例，假设有个6400M的文件，100台hadoop机器（准确地说应该是tasktracker机），默认block大小为64M，这样每台执行map的文件刚好是一个64M的block文件（假设这个分发过程已经完成，同时忽略备份数之类的细节），并且我们使用10个reduce任务来归并文件。Hadoop的mapreducer的执行过程如下：  
这100台机器上面的map都是并发、独立的执行，以wordcount为例，步骤如下：  
1、 每个map任务使用默认的textinputformat类的LineRecordReader方法按行读取文件，这个读取的行数据就被交给map函数去执行，wordcount的map做的就是提取里面的单词，并以单词为key，1为value作为输出，格式为：<wordinteger（1）>。  
2、 如果有combine，先对第一步的输出结果就行combine操作。Combine就是个小reduce操作，作用就是对某个map自己的输出结果先进行一次归并，把相同word的计数累加，这样假设某个map输出结果做如果有50%的重复word，那combine后的中间结果大小可以减少一半，可减少后续的patition、copy、sort等的开销，提高性能。  
3、 每个map对自己的输出文件进行patition操作。上面提到有10个reducer任务，那默认的patition操作就是对map的输出kay进行hash，并对10求余（hash(key)），并提供10个文件（内存足够的话可以是链表等内存数据结构），假设是r1、r2….r10这10个文件，把不同key的放到不同的文件，这次操作就可以把相同key聚合到同一个文件。由于算法一样，保证了每个map的输出结果经过这个操作后，相同key的肯定在同一个聚合文件里，比如某个单词word肯定都在r1文件里。  
4、 接下来就是copy文件的过程了，10个reducer任务各自从所有map机器上取到属于自己的文件，比如reducer1会从100台map机器上取到所有r1文件，reducer2取所有r2的文件，这样同一类word已经到了同一台reducer机器上了。  
5、 每个reducer合并（meger）自己取到的文件，reducer1就是合并100个r1文件（实际过程是在上面第4步操作中会边copy边meger，在内存中）。  
6、 合并好后进行下sort（排序）操作，再次把不同小文件中的同一个单词聚合在一起。作为提供给reduce操作的数据。  
7、 进行reduce操作，对同一个单词的value列表再次进行累加，最终得到某个单词的词频数。  
8、 Outputformat操作，把reduce结果写到磁盘。  
所以，总的流程应该是这样的：  
\* Inputformat——》map——》（combine）——》partition——》copy&merge——》sort——》reduce——》outputformat  
由此我们也可以看出，执行reduce的代价还是有些的，所以如果我们的应用只使用map就能搞定的话，那就尽量不要再有reduce操作在其中。  
from：  
http://autumnice.blog.163.com/blog/static/55520020101196941188/  
参考：  
http://blog.csdn.net/HEYUTAO007/archive/2010/07/10/5725379.aspx  
http://blog.csdn.net/wh62592855/archive/2010/07/19/5745188.aspx  
http://www.cnblogs.com/spork/archive/2010/01/11/1644342.html  
http://www.cnblogs.com/spork/archive/2010/01/11/1644346.html  
http://www.cnblogs.com/spork/archive/2010/01/11/1644350.html\[/size\]\[/color\]\[/b\]