MapReduce启动的Map/Reduce子任务简要分析

时间：2023-03-08 23:55:57
MapReduce启动的Map/Reduce子任务简要分析
对于Hadoop来说，是通过在DataNode中启动Map/Reduce java进程的方式来实现分布式计算处理的，那么就从源码层简要分析一下hadoop中启动Map/Reduce任务的过程。
首先，对于Map/Reduce端启动的任务，都是通过一些参数来控制java opts的，mapreduce.map.java.opts，mapreduce.reduce.java.opts，这些参数都在MRJobConfig类中，拿map.java.opts举例来说，org.apache.hadoop.mapred.MapReduceChildJVM类中使用了这个参数，用来构造Map或Reduce端的JVM，在下面方法中拿到了ChildJavaOpts：
private static String getChildJavaOpts(JobConf jobConf, boolean isMapTask)
如果是MapTask，JavaOpts的获得具体方法内容如下（Reduce端逻辑基本一致）：
if (isMapTask) {

      userClasspath =

          jobConf.get(

              JobConf.MAPRED_MAP_TASK_JAVA_OPTS,

              jobConf.get(

                  JobConf.MAPRED_TASK_JAVA_OPTS,

                  JobConf.DEFAULT_MAPRED_TASK_JAVA_OPTS)

          );

      adminClasspath =

          jobConf.get(

              MRJobConfig.MAPRED_MAP_ADMIN_JAVA_OPTS,

              MRJobConfig.DEFAULT_MAPRED_ADMIN_JAVA_OPTS);

    // Add admin classpath first so it can be overridden by user.

    return adminClasspath + " " + userClasspath;
userClassPath按照下面的参数顺序获得：
mapreduce.map.java.opts,

mapred.child.java.opts,

DEFAULT_MAPRED_TASK_JAVA_OPTS = "-Xmx200m”;
adminClassPath要获得的参数的基本顺序：
mapreduce.admin.map.child.java.opts，

DEFAULT_MAPRED_ADMIN_JAVA_OPTS ="-Djava.net.preferIPv4Stack=true -Dhadoop.metrics.log.level=WARN ";
最后，由于在JVM启动的参数中，后面能够覆盖掉前面的，因此userClassPath的同个选项的设置是可以覆盖adminClassPath，adminClassPath是不能保证所有参数不能被覆盖。
从源码端向上走，跳转到方法：
public static List<String> getVMCommand(InetSocketAddress taskAttemptListenerAddr, Task task, ID jvmID)
从这个方法中可以看到一个隐藏的设置，如果在选项中使用了@taskid@，是可以被替换成具体的attemptID的。
String javaOpts = getChildJavaOpts(conf, task.isMapTask());

javaOpts = javaOpts.replace("@taskid@", attemptID.toString());
代码中也以GC为例子，在注释中举例说明用法：
 // Add child (task) java-vm options.

    //

    // The following symbols if present in mapred.{map|reduce}.child.java.opts

    // value are replaced:

    // + @taskid@ is interpolated with value of TaskID.

    // Other occurrences of @ will not be altered.

    //

    // Example with multiple arguments and substitutions, showing

    // jvm GC logging, and start of a passwordless JVM JMX agent so can

    // connect with jconsole and the likes to watch child memory, threads

    // and get thread dumps.

    //

    //  <property>

    //    <name>mapred.map.child.java.opts</name>

    //    <value>-Xmx 512M -verbose:gc -Xloggc:/tmp/@taskid@.gc \

    //           -Dcom.sun.management.jmxremote.authenticate=false \

    //           -Dcom.sun.management.jmxremote.ssl=false \

    //    </value>

    //  </property>

    //

    //  <property>

    //    <name>mapred.reduce.child.java.opts</name>

    //    <value>-Xmx 1024M -verbose:gc -Xloggc:/tmp/@taskid@.gc \

    //           -Dcom.sun.management.jmxremote.authenticate=false \

    //           -Dcom.sun.management.jmxremote.ssl=false \

    //    </value>

    //  </property>

    //
一般情况下，我们在打印gc日志时，需要用-Xloggc:指定具体的目录，但是在MapReduce任务中你无法指定，因为可能两个Map会在同一台机器上执行，那样就肯定会发生gc文件覆盖，而实用@taskid@就可以避免这个问题。同样适用于OOM导致的堆栈打印避免文件意外被覆盖。
还会默认增加一个参数，用来设置java的临时文件夹（所有临时文件的建立都在这个文件夹，比如File.createTempFile）：
    vargs.add("-Djava.io.tmpdir=" + childTmpDir);
主类为：
org.apache.hadoop.mapred.YarnChild
一个YarnChild进程的最终完整命令为：
/usr/java/jdk1.7.0_11//bin/java -Djava.net.preferIPv4Stack=true -Dhadoop.metrics.log.level=WARN -Xmx2048M -Djava.io.tmpdir=/home/data5/hdfsdir/nm-local-dir/usercache/xxx/appcache/application_1413206225298_36914/container_1413206225298_36914_01_000098/tmp -Dlog4j.configuration=container-log4j.properties -Dyarn.app.container.log.dir=/home/workspace/hadoop/logs/userlogs/application_1413206225298_36914/container_1413206225298_36914_01_000098 -Dyarn.app.container.log.filesize=209715200 -Dhadoop.root.logger=INFO,CLA org.apache.hadoop.mapred.YarnChild 192.168.7.26 21298 attempt_1413206225298_36914_r_000001_0 98
从源码端来分析该命令是如何生成的：
  • $JAVA_HOME: /usr/java/jdk1.7.0_11/
  • 源码中写死: /bin/java
  • mapreduce.admin.map.child.java.opts：如果不设置，使用-Djava.net.preferIPv4Stack=true -Dhadoop.metrics.log.level=WARN；
  • mapreduce.map.java.opts: Xmx2048M;
  • 源码中添加：-Djava.io.tmpdir=/home/data5/hdfsdir/nm-local-dir/usercache/tong/appcache/application_1413206225298_36914/container_1413206225298_36914_01_000098/tmp；
  • org.apache.hadoop.mapred.MapReduceChildJVM.setLog4jProperties中设置log4j，包括日志级别，日志大小等：-Dlog4j.configuration=container-log4j.properties -Dyarn.app.container.log.dir=/home/workspace/hadoop/logs/userlogs/application_1413206225298_36914/container_1413206225298_36914_01_000098 -Dyarn.app.container.log.filesize=209715200 -Dhadoop.root.logger=INFO,CLA；
  • 主类：org.apache.hadoop.mapred.YarnChild；
  • TaskAttempt的主机地址：192.168.7.xx；
  • TaskAttempt的主机端口：212xx；
  • TaskAttempt ID：attempt_1413206225298_36914_r_000001_0；
  • JVMID：98（这是干啥的不太清楚）；
最后将进程的正常和异常输出重定向：
// Finally add the jvmID

    vargs.add("1>" + getTaskLogFile(TaskLog.LogName.STDOUT));

    vargs.add("2>" + getTaskLogFile(TaskLog.LogName.STDERR));

相关文章