Hadoop中shuffle阶段流程分析

白森

宏观上，Hadoop每个作业要经历两个阶段：Map phase和reduce phase。对于Map phase，又主要包含四个子阶段：从磁盘上读数据-》执行map函数-》combine结果 -》将结果写到本地磁盘上；对于reduce phase，同样包含四个子阶段：从各个map task上读相应的数据（shuffle）-》sort-》执行reduce函数-》将结果写到HDFS中。　　Hadoop处理流程中的两个子阶段严重降低了其性能。第一个是map阶段产生的中间结果要写到磁盘上，这样做的主要目的是提高系统的可靠性，但代价是降低了系统的性能，实际上，Hadoop的改进版–MapReduce Online去除了这个阶段，而采用其他更高效的方式提高系统可靠性（见参考资料[1]）；另一个是shuffle阶段采用HTTP协议从各个map task上远程拷贝结果，这种设计思路（远程拷贝，协议采用http）同样降低了系统性能。实际上，Baidu公司正试图将该部分代码替换成C++代码来提高性能（见参考资料[2]）。
　　本文首先着重分析shuffle阶段的具体流程，然后分析了其低效的原因，最后给出了可能的改进方法。

　　如图所示，每个reduce task都会有一个后台进程GetMapCompletionEvents，它获取heartbeat中（从JobTracker）传过来的已经完成的task列表，并将与该reduce task对应的数据位置信息保存到mapLocations中，mapLocations中的数据位置信息经过滤和去重（相同的位置信息因为某种原因，可能发过来多次）等处理后保存到集合scheduledCopies中，然后由几个拷贝线程（默认为5个）通过HTTP并行的拷贝数据，同时线程InMemFSMergeThread和LocalFSMerger会对拷贝过来的数据进行归并排序。
　　主要有两个方面影响shuffle阶段的性能：（1）数据完全是远程拷贝 （2）采用HTTP协议进行数据传输。对于第一个方面，如果采用某种策略（修改框架），让你reduce task也能有locality就好了；对于第二个方面，用新的更快的数据传输协议替换HTTP，也许能更快些, 如UDT协议（见参考资料[3]）， 它在MapReduce的另一个C++开源实现Sector/Sphere（见参考资料[4]）中被使用，效果不错！
　　————————————————————————————————————————-
　　【参考资料】
　　【1】http://code.google.com/p/hop/
　　【2】http://wenku.baidu.com/view/8225e73f0912a21614792947.html
　　【3】http://udt.sourceforge.net/
　　【4】http://sector.sourceforge.net/