(转)关于Hadoop的shuffle

 

源地址： http://blog.csdn.net/tobeandnottobe/article/details/7188747

 

我们知道每个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 round进行文件合并，每次将10个文件合并成一个。
最后5个文件可能直接进入reducer阶段。
   关于Task中所谓的Speculative Execution
是指当一个job的所有task都在running的时候，当某个task的进度比平均进度慢时才会启动一个和当前Task一模一样的任务，当其中一个 task完成之后另外一个会被中止，所以Speculative Task不是重复Task而是对Task执行时候的一种优化策略