Hadoop 之shuffle（将map输出转化为reducer的输入的过程）

Hadoop之shuffle

我们知道每个reduce task输入的key都是按照key排序的。
但是每个map的输出只是简单的key-value而非key-valuelist，所以洗牌的工作就是将map输出转化为reducer的输入的过程。

在map结束之后shuffle要做的事情：
map的输出不是简单的写入本地文件，而是更多的利用内存缓存和预排序工作，以提高效率。
io.sort.mb 用于控制 map 输出时候的内存大小，默认100Mb。
当map所使用的buffer达到一定比例的时候，会启动一个线程来将内存中数据写入磁盘。此时map过程不会暂停直到内存消耗完位置。这个线程会先将内存中的数据按照reducer的数据切分成多块,可能是按照reducer大小hash，然后对于每个块里面的数据按照key进行sort排序，此时假如定义了一个combiner函数，那么排序的结果就是combiner的输入。每当数据缓存大小达到了限制，一个新的spill文件就会被创建。所以，当map所有的数据都被处理了之后，就需要对多个spill文件进行合并操作。

combiner的作用是为了压缩mapper的输出结果，另外combiner函数需要满足n次combiner之后，输出结果都保持一致。当然，合并成一个文件的时候hadoop默认不会压缩数据，但是可以通过设置参数指定某个压缩类对数据进行压缩。

在reducer开始之前shuffle要做的事情分为两步copy和sort 阶段：

copy phrase
每个reducer  task新建几个thread用于将mapper的输出并行copy过来，copy时机是当一个mapper完成之后就可以进行。
但是reducer是如何知晓某个mapper是否完成了任务呢，mapper完成之后会给tasktracker发送一个状态更新，然后tasktraker会将该信息发送给jobtrack。然后reducer中的一个线程负责询问jobtracker 每个map的输出位置。而每个mapper上的输出数据需要等到整个job完成之后，jobtracker会通知删除。

sort phrase
将多个map输出合并成一个输入。
example：50个map输出 分5轮进行文件合并，每次将10个文件合并成一个。
最后5个文件可能直接进入reducer阶段。
关于Task中所谓的Speculative Execution是指当一个job的所有task都在running的时候，当某个task的进度比平均进度慢时才会启动一个和当前Task一模一样的任务，当其中一个task完成之后另外一个会被中止，所以Speculative Task（推测式任务）不是重复Task而是对Task执行时候的一种优化策略。


转自：http://www.cnblogs.com/zhanghuijunjava/archive/2013/04/29/3036557.html