mapreduce的执行过程

wordcount

• 关于切片
  划分maptask任务，由客户端（提交job）完成，写入文件交给mr appmaster。这里的任务切片是一个逻辑片，和hdfs的block是不同的概念。对于大文件，切片一般按照block的大小来切，方便后续执行maptask时数据的读取。对于一些小文件，需要把多个block切成一片。一个切片对应一个maptask实例。切片由fileinputformat中的getSplit()来完成，结果写入切片规划文件。切片大小（由minsize，maxsize，blocksize几个值决定）的逻辑：Math.max(minSize,Math.min(maxSize,blockSize))。参数在conf中设置。

• 关于mapreduce中的序列化
  map和reduce的输出都需要经过传输和实体化存储，所以都需要让它实现一个序列化框架，hadoop里面有一个内置的序列化框架writable。输出的所有数据都需要实现writable接口。如果输出不是它内置的数据类型（比如text，longwritable，intwritrable），而是我们一个自定义的对象，就必须去实现writable接口。

• mapreduce全过程
  mapreduce中通过InputFormat组件来读取数据。我们读入的数据是什么，后续处理的数据时什么完全取决于这个接口。它把外部的数据源与内部的处理逻辑完全解耦。InputFormat默认读取文本文件的组件是：TextInputformat。maptask读取数据时我们调用其成员RecordReader的read()方法，取得kv对返回给Mapper中的map()方法。

InputFormat

map()方法中是我们自定义的业务逻辑，执行完逻辑后context.write(k,v)。把map过后的结果交给Outputcollector。OutputCollector把数据写入环形缓冲区（就是一个环形数组）。环形缓冲区中有一个保留区来进行分区排序。
分区排序实现：
默认分区是：HashPartitioner（hashcode模除以2(这个数字可以用来调整分区数)）。排序：quicksort和外部排序混合使用。Key.compareTo。
缓冲区中的数据积累到一定阈值（如80%）通过spiller溢出（maptask完毕后还会有一次清空溢出）。溢出一次写一个文件，这些文件分区且有序。最后再merge这些文件（mergesort）形成一个maptask的最终文件。在这个文件中有几个分区就会有几个reducer，且这个分区与环形缓冲区中分区相对应。

shuffle

maptask结束后，开始reducetask。reducer到执行maptask的机器下载属于自己分区的数据到其本地磁盘工作目录，有几个maptask对应下载到几个文件。然后将这几个文件merge（mergesort）。reducer里自定义的reduce方法，对每一组key（相同key即为一组，通过GroupingComparator(k,nextk)组件判断key是否相同）。reduce(k,v)中一组中第一个传入的key作为k，value是一个可以迭代这一组value的迭代器。执行完自定义reduce()的逻辑后context.write(k,v)。然后调用OutputFormat（默认TextOutputFormat）中的RecordWriter中的write(k,v)方法把结果写入FileoutputFormat.setoutputpath(file://'xx')中。不同的reducer写出不同的文件，part-r00000，part-r00001……。

• combiner
  在maptask的本地先完成一些类似reduce的操作。从环形缓冲区溢出时spiller去调用combiner把相同key的value累加。在merge溢出文件时也可以调用combiner累加value。