Hadoop Shuffle过程分析

yzwj2004

hadoop shuffle 优化

在hadoop中，在map/reduce的shuffle阶段，jetty用于数据传输。提高map/reduce的效率，针对shuffle的优化也是很重要的。它可以在以下几个方面进行优化：

• 优化jetty
  
• 减少map输出
  
• 用netty来替换jetty
  
• 压缩传输
  

Hadoop Map阶段的输出机制

一个作业由Map Task和Reduce Task组成。通常Map Task的输出，当作Reduce Task的输入。如果该作业没有Reduce Task时，那么该Job的输出目的地为HDFS。在MapTask类中有一个run方法。

@Override
public void run(final JobConf job, final TaskUmbilicalProtocol umbilical)
throws IOException, ClassNotFoundException, InterruptedException {
this.umbilical = umbilical;  // 与tasktracker通信的代理
// start thread that will handle communication with parent
//起一个新的线程reporter去调用umbilical与父节点通信
TaskReporter reporter = new TaskReporter(getProgress(), umbilical,
jvmContext);
reporter.startCommunicationThread();
//检查是否是新api
boolean useNewApi = job.getUseNewMapper();
//初始化map的上下文环境
initialize(job, getJobID(), reporter, useNewApi);
// check if it is a cleanupJobTask
if (jobCleanup) {
runJobCleanupTask(umbilical, reporter);
return;
}
if (jobSetup) {
runJobSetupTask(umbilical, reporter);
return;
}
if (taskCleanup) {
runTaskCleanupTask(umbilical, reporter);
return;
}
if (useNewApi) {
//新的api是位于org.apache.hadoop.mapreduce下面的包
runNewMapper(job, splitMetaInfo, umbilical, reporter);
} else {
//新的api是位于org.apache.hadoop.mapred下面的包
runOldMapper(job, splitMetaInfo, umbilical, reporter);
}
done(umbilical, reporter);
}
 在新api中的输出分两种情况：

• 如果没有Reduce，那些就通常直接输出到hdfs上,相应地址由mapred.output.dir控制,out为NewDirectOutputCollector； 
  
• 如果有Reduce,out为NewOutputCollector,它有一定大小的缓存区kvbuffer，输出的数据会先写到这个缓存里面。这里面有复杂的内存控制机制，需要相关参数来协调，如：io.sort.mb(默认<4096),io.sort.record.percent(default: 0.05),io.sort.spill.percent(default: 0.8)等。io.sort.mb的值不能大于2^12=4096。kvbuffer大小的计算公式如下:( io.sort.mb * (2^20) ) * ( 1 - io.sort.record.percent )，分布到不同partition的kv都会存到这个kvbuffer里面，同时通过两个数组kvoffsets，kvindices来建立索引。如果kvbuffer快要满时，就会spill和写到本地磁盘，存取的格式是IFile格式：<key-len, value-len, key, value> 。
  

 Shuffer机制

待续... 

Jetty 6.1.14和6.1.26版本比较

近段时间对jetty 6.1.14和jetty 6.1.26进行了一次简单的性能测试（该测试是通过jetty server下载5G的文件），发现前者的传输性能明显优于后者，是后者的三倍左右。jetty 6.1.14在测试过程往往可以达到网卡的最大速度，而jetty 6.1.26就要打3折。在new一个jetty server时可以配置一个bufferSize，这个bufferSize的配置也对它们的性能有一定的影响，根据业务配置一个合适的值是能够提高一定的性能的。
在Jetty和Hadoop的Jira上相继报道了相关bug, 其中就包括：

• jetty 6.1.14很容易OOM，升级到jetty 6.1.26就没有这个问题, JETTY-1374，
  
• jetty 6.1.26很容易造成fetch failure等问题 HADOOP-2980,HADOOP-2389，
  

上述两个问题很来严重，因此jetty的开发者都建议hadoop 依赖的jetty包升级到7.0以上。
 

PS：

1. 从hadoop日志里经常可以看到以下的输出日志：
2013-05-28 20:07:05,476 INFO org.mortbay.log: org.mortbay.io.nio.SelectorManager$SelectSet@3160e069 JVM BUG(s) - injecting delay35 times
2013-05-28 20:07:05,476 INFO org.mortbay.log: org.mortbay.io.nio.SelectorManager$SelectSet@3160e069 JVM BUG(s) - recreating selector 35 times, canceled keys 561 times

　　这是JVM 1.6的一个bug: Selector.select(timeout)时无法block住和select时出现句柄存在(FileExistException)的异常，该bug在jetty中已经作了workround (在代码 org.mortbay.io.nio.SelectorManager中可以查看到)，同时JVM 1.7也已经把它修复了。
　　2. 关于Java OOM时如何查看Heap相关信息。

• 可以通过给java加下面这个参数可以在OOM时把Heap信息dump出来：
  

-XX:+HeapDumpOnOutOfMemoryError

•  在OOM之前，还可以通过jmap命令导出来：
  

jmap -dump:format=b,file=<filename.hprof> <pid>

Netty vs Jetty
　　待续...