Hadoop如何组织中间数据的存储和传输(源码级分析)1

lixiaolong

　　Hadoop以可扩展、易用、分布式处理海量数据为目标，在海量数据处理领域不断地制造着神话。其中，最为重要的一个特性就是中间数据的使用上。Hadoop将Map阶段产生的结果，不直接存入HDFS,而是放在本地磁盘中作为中间数据存储起来。等到Reduce启动以后，就从Map阶段拉取中间数据。这个过程成为了MapReduce中的一个大家津津乐道的经典过程，但是，它内部是如何实现的呢？
传输其中中间是通过Http方式来传输Map阶段产生的中间文件到Reducer，分析这个过程是如何实现的，首先看Hadoop的执行task的流程。

这里面最引人注意的是Hadoop重新启动了一个Java虚拟机来启动一个Task，这个task可能是MapTask和ReduceTask，这样做是为了用户定义的MapTask和ReduceTask与JobTracker-TaskTracker体系隔离开来，保证安全。另外也使得一些配置参数可以重新设置。
下面貼上一段我分析出来Child的执行参数：

HADOOP_CLIENT_OPTS:-Dhadoop.tasklog.taskid=attempt_201105201525_0001_m_000002_0 -Dhadoop.tasklog.iscleanup=false -Dhadoop.tasklog.totalLogFileSize=0
HADOOP_TOKEN_FILE_LOCATION:/tmp/hadoop-klose/mapred/local/taskTracker/klose/jobcache/job_201105201525_0001/jobToken
HADOOP_ROOT_LOGGER:INFO,TLA
LD_LIBRARY_PATH:/tmp/hadoop-klose/mapred/local/taskTracker/klose/jobcache/job_201105201525_0001/attempt_201105201525_0001_m_000002_0/work:/opt/sun-jdk-1.6.0.17/jre/lib/i386/server:/opt/sun-jdk-1.6.0.17/jre/lib/i386:/opt/sun-jdk-1.6.0.17/jre/../lib/i386
klose: shexec getExecString:
bash
-c
 export JVM_PID=`echo $$` ; 'ulimit' '-v' '-1' ;exec setsid '/opt/sun-jdk-1.6.0.17/jre/bin/java' '-Djava.library.path=/home/klose/hadoop-0.21.0/bin/../lib/native/Linux-i386-32:/tmp/hadoop-klose/mapred/local/taskTracker/klose/jobcache/job_201105201525_0001/attempt_201105201525_0001_m_000002_0/work' '-Xmx1024m' '-Djava.io.tmpdir=/tmp/hadoop-klose/mapred/local/taskTracker/klose/jobcache/job_201105201525_0001/attempt_201105201525_0001_m_000002_0/work/tmp' '-classpath' '/home/klose/hadoop-0.21.0/bin/../conf:/opt/sun-jdk-1.6.0.17/lib/tools.jar:/home/klose/hadoop-0.21.0/bin/..:/home/klose/hadoop-0.21.0/bin/../hadoop-common-0.21.0.jar:/home/klose/hadoop-0.21.0/bin/../hadoop-common-test-0.21.0.jar:/home/klose/hadoop-0.21.0/bin/../hadoop-hdfs-0.21.0-sources.jar:/home/klose/hadoop-0.21.0/bin/../hadoop-hdfs-0.21.0.jar:/home/klose/hadoop-0.21.0/bin/../hadoop-hdfs-ant-0.21.0.jar:/home/klose/hadoop-0.21.0/bin/../hadoop-hdfs-test-0.21.0-sources.jar:/home/klose/hadoop-0.21.0/bin/../hadoop-hdfs-test-0.21.0.jar:/home/klose/hadoop-0.21.0/bin/../hadoop-mapred-0.21.0-sources.jar:/home/klose/hadoop-0.21.0/bin/../hadoop-mapred-0.21.0.jar:/home/klose/hadoop-0.21.0/bin/../hadoop-mapred-examples-0.21.0.jar:/home/klose/hadoop-0.21.0/bin/../hadoop-mapred-test-0.21.0.jar:/home/klose/hadoop-0.21.0/bin/../hadoop-mapred-tools-0.21.0.jar:/home/klose/hadoop-0.21.0/bin/../lib/ant-1.6.5.jar:/home/klose/hadoop-0.21.0/bin/../lib/asm-3.2.jar:/home/klose/hadoop-0.21.0/bin/../lib/aspectjrt-1.6.5.jar:/home/klose/hadoop-0.21.0/bin/../lib/aspectjtools-1.6.5.jar:/home/klose/hadoop-0.21.0/bin/../lib/avro-1.3.2.jar:/home/klose/hadoop-0.21.0/bin/../lib/commons-cli-1.2.jar:/home/klose/hadoop-0.21.0/bin/../lib/commons-codec-1.4.jar:/home/klose/hadoop-0.21.0/bin/../lib/commons-el-1.0.jar:/home/klose/hadoop-0.21.0/bin/../lib/commons-httpclient-3.1.jar:/home/klose/hadoop-0.21.0/bin/../lib/commons-lang-2.5.jar:/home/klose/hadoop-0.21.0/bin/../lib/commons-logging-1.1.1.jar:/home/klose/hadoop-0.21.0/bin/../lib/commons-logging-api-1.1.jar:/home/klose/hadoop-0.21.0/bin/../lib/commons-net-1.4.1.jar:/home/klose/hadoop-0.21.0/bin/../lib/core-3.1.1.jar:/home/klose/hadoop-0.21.0/bin/../lib/ftplet-api-1.0.0.jar:/home/klose/hadoop-0.21.0/bin/../lib/ftpserver-core-1.0.0.jar:/home/klose/hadoop-0.21.0/bin/../lib/ftpserver-deprecated-1.0.0-M2.jar:/home/klose/hadoop-0.21.0/bin/../lib/hsqldb-1.8.0.10.jar:/home/klose/hadoop-0.21.0/bin/../lib/jackson-core-asl-1.4.2.jar:/home/klose/hadoop-0.21.0/bin/../lib/jackson-mapper-asl-1.4.2.jar:/home/klose/hadoop-0.21.0/bin/../lib/jasper-compiler-5.5.12.jar:/home/klose/hadoop-0.21.0/bin/../lib/jasper-runtime-5.5.12.jar:/home/klose/hadoop-0.21.0/bin/../lib/jdiff-1.0.9.jar:/home/klose/hadoop-0.21.0/bin/../lib/jets3t-0.7.1.jar:/home/klose/hadoop-0.21.0/bin/../lib/jetty-6.1.14.jar:/home/klose/hadoop-0.21.0/bin/../lib/jetty-util-6.1.14.jar:/home/klose/hadoop-0.21.0/bin/../lib/jsp-2.1-6.1.14.jar:/home/klose/hadoop-0.21.0/bin/../lib/jsp-api-2.1-6.1.14.jar:/home/klose/hadoop-0.21.0/bin/../lib/junit-4.8.1.jar:/home/klose/hadoop-0.21.0/bin/../lib/kfs-0.3.jar:/home/klose/hadoop-0.21.0/bin/../lib/log4j-1.2.15.jar:/home/klose/hadoop-0.21.0/bin/../lib/mina-core-2.0.0-M5.jar:/home/klose/hadoop-0.21.0/bin/../lib/mockito-all-1.8.2.jar:/home/klose/hadoop-0.21.0/bin/../lib/oro-2.0.8.jar:/home/klose/hadoop-0.21.0/bin/../lib/paranamer-2.2.jar:/home/klose/hadoop-0.21.0/bin/../lib/paranamer-ant-2.2.jar:/home/klose/hadoop-0.21.0/bin/../lib/paranamer-generator-2.2.jar:/home/klose/hadoop-0.21.0/bin/../lib/qdox-1.10.1.jar:/home/klose/hadoop-0.21.0/bin/../lib/servlet-api-2.5-6.1.14.jar:/home/klose/hadoop-0.21.0/bin/../lib/slf4j-api-1.5.11.jar:/home/klose/hadoop-0.21.0/bin/../lib/slf4j-log4j12-1.5.11.jar:/home/klose/hadoop-0.21.0/bin/../lib/xmlenc-0.52.jar:/home/klose/hadoop-0.21.0/bin/../lib/jsp-2.1/*.jar:/home/klose/hadoop-0.21.0/bin/../hdfs/conf:/home/klose/hadoop-0.21.0/bin/../hdfs/hadoop-hdfs-*.jar:/home/klose/hadoop-0.21.0/bin/../hdfs/lib/*.jar:/home/klose/hadoop-0.21.0/mapred/bin/../conf:/home/klose/hadoop-0.21.0/mapred/bin/../hadoop-mapred-*.jar:/home/klose/hadoop-0.21.0/mapred/bin/../lib/*.jar:/home/klose/hadoop-0.21.0/mapred/bin/../hadoop-mapred-*.jar:/home/klose/hadoop-0.21.0/mapred/bin/../lib/*.jar:/tmp/hadoop-klose/mapred/local/taskTracker/klose/jobcache/job_201105201525_0001/jars/classes:/tmp/hadoop-klose/mapred/local/taskTracker/klose/jobcache/job_201105201525_0001/jars/job.jar:/tmp/hadoop-klose/mapred/local/taskTracker/klose/jobcache/job_201105201525_0001/attempt_201105201525_0001_m_000002_0/work' '-Dhadoop.log.dir=/home/klose/hadoop-0.21.0/bin/../logs' '-Dhadoop.root.logger=INFO,TLA' '-Dhadoop.tasklog.taskid=attempt_201105201525_0001_m_000002_0' '-Dhadoop.tasklog.iscleanup=false' '-Dhadoop.tasklog.totalLogFileSize=0' 'org.apache.hadoop.mapred.Child' '127.0.0.1' '50788' 'attempt_201105201525_0001_m_000002_0' '/home/klose/hadoop-0.21.0/bin/../logs/userlogs/job_201105201525_0001/attempt_201105201525_0001_m_000002_0' '1522055976'  < /dev/null  1>> /home/klose/hadoop-0.21.0/bin/../logs/userlogs/job_201105201525_0001/attempt_201105201525_0001_m_000002_0/stdout 2>> /home/klose/hadoop-0.21.0/bin/../logs/userlogs/job_201105201525_0001/attempt_201105201525_0001_m_000002_0/stderr

　　分析：
1）org.apache.hadoop.mapred.Child是MapTask和ReduceTask的启动类；
2）使用命令来启动了一个新的JVM。setsid run a program in a new session.这样使得程序就从主程序中独立出来。
3）在JobTracker和TaskTracker一个job会有一个Secret Key的JobToken，辨识task。当然Map阶段产生的中间数据，会通过token的secret来进行加密，保证只有含有同一secret的reduce阶段可以获取中间数据。

讲到这里，就必须介绍一下Hadoop中间数据的操作过程了，由于Hadoop的MapReduce的编程框架与运行时环境是高耦合的。所以，这个运行的过程在org.apache.hadoop.mapred.MapTask 和 org.apache.hadoop.mapred.ReduceTask中实现。
中间数据的拷贝存在与MapReduce的Shuffle阶段(也叫洗牌阶段)


分析Map端过程：
1）缓冲区(io.sort.mb --- 缓冲区大小， io.sort.spill.percent --- 缓冲区向本地磁盘spill的时机)，在达到缓冲区的threshold的时刻，后台线程开始把内容写入本地磁盘中。同时map会继续写到缓冲区，但如果缓冲区被填满，map会阻塞至到溢出过程结束。
tips：用户在继承了Mapper，写了map函数，如何与MapTask进行交互的呢？
  a)在经历了一番心跳和调度之后，TaskTracker获取了一个任务。这个过程详细的解读：hadoop情景调度1 和hadoop情景调度2 .
  b)TaskTracker启动launchTask，然后会通过TaskRunner线程启动JVM，launchJvmAndWait()会通过JVMManager启动一个Child JVM，具体操作使用了reapJvm(),由于conf参数中可以设置JVM复用，在reapJvm会首先进行一系列的判断操作，当条件都不满足的情况下，就启动一个新的JVM。spawnNewJvm启动JvmRunner。
  c)JvmRunner启动一个Child,Child的启动参数如上面的贴出的代码所示。注意setsid的使用。
  d)启动一个Child Process,其实经常报Java Heap space的问题，可以设置mapred.child.java.opts的值。
  e)启动以后通过MapTask.run(JobConf, TaskUmbilicalProtocol)来启动任务，这里TaskUmbilicalProtocol的实例用来与TaskTracker进程进行任务状态交互。
  f)MapTask和ReduceTask是我们需要重点研究的类，它在控制实现map、reduce的操作过程。
 


2）MapTask通过反射获取用户定义MapperClass的map(K,V,Context)，并将其它Context环境准备就绪。
　　3）context.write(K,V)的执行流程。
　　a) Context 通过参数传递，被实例化MapperContext一个实现类的对象；
　　b)MapperContext的实现类MapperContextImpl类是TaskInputOutputContextImpl的子类，在类定义过程中，将output从MapTask传递到，TaskInputOutputContextImpl中，TaskInputOutputContext的write(K,V)的方法，使用output.write(K,V)。这里的output实际上是MapTask中的NewDirectOutputCollector或者NewOutputCollector的一个。
　　c)以NewOutputCollector为例，write(K,V)使用了MapOutputBuffer的collect方法。MapOutputBuffer是MapTask的一个内部类，它解答了这篇Blog的一半的内容，Hadoop是如何组织中间数据从内存到磁盘的过程的。下面将重点介绍这个类。