Hadoop MapReduce 初步学习总结

 

在Hadoop中一个作业被提交后，其后具体的执行流程要经历Map任务的提交中间结果处理，Reduce任务的分配和执行直至完成这些过程，下面就是MapReduce中作业详细的执行流程图（摘自《Hadoop实战》）。

 

 

 

在整个过程中，客户端中，编写MapReduce代码，配置并提交作业；JobTracker中负责初始化作业，分配作业，与TaskTracker进行通信，协调整个作业的运行；TaskTracker要保持与JobTracker的通信，在分配的数据分片上执行Map或Reduce任务；而在整个过程中，HDFS保存作业的数据，配置信息等，以及作业的执行结果。

  

MapReduce确保每个reducer的输入都是按键排序的，系统执行排序的过程。将map输出作为输入传给reduce，这个过程称为shuffle。shuffle过程属于不断被优化和改进的部分。

 

Hadoop的版本为2.2.0，不同版本的参数名称会稍有差异。

 

Map端

 

map函数开始产生输出时并不是简单第写出到磁盘，它利用缓冲的方式写到内存，并处于效率的考虑进行预排序。

 

每个map任务都有一个环形缓冲区用于存储任务的输出，缓冲区的大小为mapreduce.task.io.sort.mb属性的值，一旦缓冲区达到阈值，就会启动一个后台线程开始将内容写到磁盘中,这部分内存是占用map端的JVM堆的。阈值是一个百分比（mapreduce.map.sort.spill.percent），表示当环形缓冲区中的数据达到这个百分比之后，就开始写磁盘了。在写磁盘的过程中，map输出继续被写到缓冲区中，但是如果写磁盘的速度非常慢，就会导致map被阻塞。

 

在写磁盘之前，线程首先根据数据最终要传送到的reducer将数据进行分区(partition)，比如如果手动设置reducer的数量为64，那么spill文件中就会有64个分区，每个分区中的数据都是排序好的。

 

只要内存缓冲达到mapreduce.map.sort.spill.percent中设置的值，就会spill一个文件。这意味着在map任务执行完成后，也许有多个溢出文件，这些文件需要最终被合并成一个已分区且已排序的输出文件。这个合并过程是一个多路归并排序的过程，至于一次最多能够合并多少个文件，可以使用属性mapreduce.task.io.sort.factor配置，这个属性使用的范围比较广，以后reduce端数据的合并仍然是这个属性控制。当spill文件归并完成后，Map就会删除所有的临时spill文件，并告诉TaskTracker任务已经完成。

 

 

Reduce端

map的输出文件位于运行map任务的tasktracker上的本地磁盘（注意，非HDFS上，这部分数据也没有必要存储到HDFS上）。而对于reduce任务，需要集群上的多个map任务（我认为合理的话，应该是每个map输出）的map输出作为其特殊的分区文件。每个map任务的完成时间可能会有所不同，因此只要有一个任务完成，reduce任务就开始复制其输出。

 

map执行的每个tasktracker是负责提供map中间结果的服务器，使得reducer可以通过HTTP的方式，与map端进行通信。用于提供map端数据复制服务的工作线程数量是由mapreduce.tasktracker.http.threads来指定的，这是一个针对每个tasktracker而不是单个job的参数。

 

reduce任务的复制阶段，reduce端会有少量复制线程，能够并行获得map输出，这可以通过mapreduce.reduce.shuffle.parallelcopies参数进行设置。

 

复制过来的map数据首先会被缓存到Reduce端的内存缓冲区中，缓冲区的大小由mapreduce.reduce.shuffle.input.buffer.percent控制，这是一个百分比，指示占有reduce端JVM堆内存的百分比。随着数据的增大，同样也会进行输出磁盘的工作，后台线程会将它们合并为更大的排序后文件。

 

最后阶段，Reduce端的内存缓冲区数据将会被释放，最终合并成一个文件，但Reduce端仍然会在内存中保留部分map输出结果，保留的百分比是由属性mapreduce.reduce.input.buffer.percent，这也是一个百分比，表示占用Reduce端JVM堆内存的百分比，默认为0.0，意味着不会进行缓存。

 

进行Reduce阶段，直接将整理过后的数据输入reduce函数。

 

 

关于Combiner

当map操作输出其键值对时，都已经在内存中存在了。为了性能的考虑，有时候提供一个拥有reduce功能的Combiner是非常有效的。如果由combiner，map的键值对不会马上被写出到磁盘中，而是先被收集到list中，一个key值一个list（类似MapList的数据结构）。

 

Combiner函数可能会在map的merge操作完成之前，也可能在merge之后执行，这个时机由配置参数min.num.spill.for.combine指定(该值默认为3)，也就是说在map端产生的spill文件最少有min.num.spill.for.combine的时候，Combiner函数会在merge操作合并最终的本机结果文件之前执行，否则在merge之后执行。通过这种方式，就可以在spill文件很多并且需要做conbine的时候，减少写入本地磁盘的数据量，同样也减少了对磁盘的读写频率，可以起到优化作业的目的。

 

hadoop文档中也有说明Combiner可能被执行也可能不被执行，如果当前集群在很繁忙的情况下job就是设置了也不会执行Combiner。

 

需要注意的是，虽然combiner使用合适可以提高Job执行作业的吞吐量，但不合适的应用场景可能导致输出结果不正确。Combiner的输出是Reducer的输入，绝不能改变最终的计算结果。

 

运行时参数调整

 

修改job配置

1. 在执行hadoop jar时，可以通过-D参数的方式配置虚拟机参数，
   • mapreduce.map.java.opts指示配置map端的参数
   • mapreduce.reduce.java.opts指示配置reduce端的参数
2. 注意：这两个参数必须要配置在Main函数的其他参数之前，并加上-D, 否则不起作用。

hadoop jar **1-SNAPSHOT.jar com.xxx.Main -Dmapreduce.map.java.opts=-Xmx2048M -Dmapreduce.reduce.java.opts=-Xmx2048M /xxx.parameter.json

 

在oozie中修改配置

在oozie中通过java action方式执行时，也可以通过-D参数的方式配置虚拟机参数

注意：通过测试发现<java-opts>参数未起作用。

 

<action name="test" retry-max="3" retry-interval="1">  
        <java>  
            <job-tracker>${jobTracker}</job-tracker>  
            <name-node>${namenode}</name-node>  
            <main-class>com.miaozhen.yo.tcpreporter.Main</main-class>  
            <java-opts>-Xmx4096M</java-opts>  
            <arg>-Dmapreduce.map.java.opts=-Xms1024M -Xmx2048M</arg>  
            <arg>-Dmapreduce.reduce.java.opts=-Xms1024M -Xmx2048M</arg>  
            <arg>${conf}</arg>  
            <arg>test</arg>  
        </java>  
        <ok to="end"/>  
        <error to="end"/>  
    </action>  

 

 

 

总体优化

对于整个过程总的原则是要给Shuffle过程尽可能多提供空间，同时需要确保map函数和reduce函数能得到足够的内存使用。map和reduce函数都需要尽量少地使用内存。

 

map和reduce任务设置的java执行参数为：mapreduce.map.java.opts和mapreduce.reduce.java.opts，可以在上面设置对应的Java堆，永久代等大小和其他Java参数。

 

在map端，应该避免多次溢出写磁盘，如果能够估算出map输出，可以合理地设置io.sort.mb的值。

 

在reduce端，中间数据全部驻留在内存中时能够获得最佳性能。默认情况下，所有内存都预留给reduce函数了（mapreduce.reduce.input.buffer.percent被设置成0.0），但如果reduce函数的内存需求不太大，就可以相应地将数据缓存。