基于Hadoop系统的MapReduce数据流优化

1  Hadoop管道改进思想
    在Hadoop系统的实现中，Map端的输出数据首先被溢写入本地磁盘，当本机任务完成后通知JobTracker，然后Reduce端在得到JobTracker的通知后会发出HTTP请求，利用复制的方式从相应的Map端拉回其输出。这样的方式只能等该Map任务完成后才能开始执行Reduce任务，并且Map任务和Reduce任务的执行是分离的。
    我们的改进思想是使Map任务和Reduce任务能够以管道的方式执行，即Map任务开始产生输出后直接发送给相应的Reduce任务，这需要在用户提交作业后JobTracker就分配相应的Map任务和Reduce任务，并将每个Map任务的位置发送给Reduce任务。每个Reduce任务启动后就联系每个任务并打开一个Socket。当每个Map输出一产生后Mapper便决定发送的分区（Reduce任务）并通过适合的Socket直接发送。
    Reduce任务接收从每个Map任务收到的管道数据并将其存储在内存缓冲区中，在需要时将已排好序的缓冲区内容写到磁盘。一旦Reduce任务得知每个Map任务均已完成，它执行已排序内容的最终合并，然后调用用户自定义的Reduce函数，将输出写入HDFS。
2  面临问题及解决方法
在以上的改进思想中面临以下几个实际问题，我们将对其进行分析并提出解决方法。
(1) Hadoop系统可能没有最够可用任务槽来调度作业的每个任务。
由于任务槽的限制，可能部分Reduce还没有被调度，则Map输出无法直接发送给它。改进方法为将这部分Map的输出写入磁盘，当Reduce任务被分配任务槽后，就像Hadoop系统一样从Map任务复制数据。
(2) 打开每个Map任务和Reduce任务之间的Socket需要大量的TCP连接。
大量的TCP将占用过多的网络带宽，容易造成网络堵塞。为了减少并发TCP连接数，每个Reducer可以被配置为从有限数量的Mapper以管道方式传送数据，其余的数据按Hadoop系统的传统方式从Mapper拉回。
(3) 调用Map函数和将输出写入管道Socket使用相同的线程。
这可能将导致如下情况，由于Reducer过度使用而造成网络I/O堵塞，则Mapper也无法做有用的工作。改进方法为以单独线程运行Map函数，将其输出存储到内存缓冲区，然后由另一线程定期发送缓冲区内容到管道Reducer端。
(4) Map任务急切发送产生的数据，阻止了Combiner的使用，并将一些排序工作从Mapper移动到Reducer。
    Map任务一产生数据便使用管道方式传送给对应的Reduce而没有机会应用Combiner函数，会增大网络流量。同时Map的排序过程更多的转移到Reduce任务会Reduce的响应时间并带来很大的系统开销，因为通常Map任务的数量远远大于Reduce任务。改进方法为修改内存缓冲区设计，不直接发送缓冲区内容给Reducer，而是等到缓冲区增长到阈值大小后应用Combiner函数，按分区和key值进行排序后将内容溢写入磁盘。第二个线程监测溢写文件并将其以管道方式发送给Reducer。如果Reducer能够赶上Mapper并且网络不是瓶颈，则溢写文件在产生后马上发送给Reducer。否则溢写文件将逐渐增加，Mapper定期对其应用Combiner函数将多个溢写文件合并成一个单一的大型文件。
3  改进系统的实现
    UC Berkeley的Tyson Condie等人基于MapReduce Online论文实现了Hadoop Online Prototype(HOP)[13]系统，它除了能够实现作业内Mapper到Reducer的管道之外，还利用“快照”技术实现了作业间Reducer到Mapper的管道执行。HOP还支持连续查询，这使得MapReduce程序能够被用于事件监测和流处理的等实时应用。同时，HOP保留了Hadoop的容错特性，并能够运行未修改的用户定义的MapReduce程序。
    HOP实现的数据流与Hadoop系统的对比如下图所示：



    在Hadoop-0.19.2中，org.apache.hadoop.mapred包实现了Hadoop MapReduce思想，HOP增加了org.apache.hadoop.mapred.bufmanager包来管理Map和Reduce任务的输入和输出。其中主要包含的类如下表所示：



    使用HOP系统在伪分布式上能够成功运行MapReduce作业，但是在集群中部署后执行WordCount应用程序时，当Map阶段完成后Reduce阶段完成25%时，作业长时间停滞无法继续执行，显示如下图所示的错误：



    我们参考HOP程序对Hadoop-0.19.2进行修改，并使用Ant编译，成功后执行情况与HOP相同，同样在集群情况下执行MapReduce作业过程中停滞。分析原因，如果HOP系统本身的实现不存在问题，那可能是实验集群的配置或者网络存在问题，但是具体原因一直没有发现并解决。
    基于Hadoop系统优化的测试实验使用HOP系统进行，能够使Map过程和Reduce过程管道执行。根据MapReduce Online论文中所示，作者在Amazon EC2上使用60节点集群执行了性能测试实验，对从维基百科提取的5.5GB数据进行排序，结果如下图所示：



    实验结果显示HOP比Hadoop更有优势，大大减少了作业完成时间，具有更高的系统利用率。
    但是由于在集群中执行HOP出现错误，为了验证其优化效果，使用伪分布式执行WordCount程序，通过与Hadoop系统上执行原始程序进行对比，得到两者的执行时间分别为314秒（HOP）和266秒（Hadoop），两者的Map过程和Reduce过程分别如下图1和下图2所示。通过对比可以发现，HOP系统确实实现了Map过程和Reduce过程的管道执行，但是在作业执行时间上HOP系统更长，这于上图的对比分析图有差异。具体可能由HOP系统实现、集群数量及配置、处理数据量等多方面原因导致。但是HOP这种改进思想还是很值得学习和借鉴。