MapReduce V1：Job提交流程之JobClient端分析

我们基于Hadoop 1.2.1源码分析MapReduce V1的处理流程。

MapReduce V1实现中，主要存在3个主要的分布式进程（角色）：JobClient、JobTracker和TaskTracker，我们主要是以这三个角色的实际处理活动为主线，并结合源码，分析实际处理流程。下图是《Hadoop权威指南》一书给出的MapReduce V1处理Job的抽象流程图：



如上图，我们展开阴影部分的处理逻辑，详细分析Job提交在JobClient端的具体流程。
在编写好MapReduce程序以后，需要将Job提交给JobTracker，那么我们就需要了解在提交Job的过程中，在JobClient端都做了哪些工作，或者说执行了哪些处理。在JobClient端提交Job的处理流程，如下图所示：



上图所描述的Job的提交流程，说明如下所示：

  1. 在MR程序中创建一个Job实例，设置Job状态
  2. 创建一个JobClient实例，准备将创建的Job实例提交到JobTracker
  3. 在创建JobClient的过程中，首先必须保证建立到JobTracker的RPC连接
  4. 基于JobSubmissionProtocol协议远程调用JobTracker获取一个新的Job ID
  5. 根据MR程序中配置的Job，在HDFS上创建Job相关目录，并将配置的tmpfiles、tmpjars、tmparchives，以及Job对应jar文件等资源复制到HDFS
  6. 根据Job配置的InputFormat，计算该Job输入的Split信息和元数据（SplitMetaInfo）信息，以及计算出map和reduce的个数，最后将这些信息连通Job配置写入到HDFS（保证JobTracker能够读取）
  7. 通过JobClient基于JobSubmissionProtocol协议方法submitJob提交Job到JobTracker

MR程序创建Job

下面的MR程序示例代码，已经很熟悉了：

01 public static void main(String[] args) throws Exception {
02 Configuration conf = new Configuration();
03 String[] otherArgs = new GenericOptionsParser(conf, args).getRemainingArgs();
04 if (otherArgs.length != 2) {
05 System.err.println("Usage: wordcount <in> <out>");
06 System.exit(2);
07 }
08 Job job = new Job(conf, "word count");
09 job.setJarByClass(WordCount.class);
10 job.setMapperClass(TokenizerMapper.class);
11 job.setCombinerClass(IntSumReducer.class);
12 job.setReducerClass(IntSumReducer.class);
13 job.setOutputKeyClass(Text.class);
14 job.setOutputValueClass(IntWritable.class);
15 FileInputFormat.addInputPath(job, new Path(otherArgs[0]));
16 FileOutputFormat.setOutputPath(job, new Path(otherArgs[1]));
17 System.exit(job.waitForCompletion(true) ? 0 : 1);
18 }

在MR程序中，首先创建一个Job，并进行配置，然后通过调用Job的waitForCompletion方法将Job提交到MapReduce集群。这个过程中，Job存在两种状态：Job.JobState.DEFINE和Job.JobState.RUNNING，创建一个Job后，该Job的状态为Job.JobState.DEFINE，Job内部通过JobClient基于org.apache.hadoop.mapred.JobSubmissionProtocol协议提交给JobTracker，然后该Job的状态变为Job.JobState.RUNNING。

Job提交目录submitJobDir

通过如下代码可以看到，Job提交目录是如何创建的：

1 JobConf jobCopy = job;
2 Path jobStagingArea = JobSubmissionFiles.getStagingDir(JobClient.this, jobCopy); // 获取到StagingArea目录
3 JobID jobId = jobSubmitClient.getNewJobId();
4 Path submitJobDir = new Path(jobStagingArea, jobId.toString());

获取StagingArea目录，JobClient需要通过JobSubmissionProtocol协议的远程方法getStagingAreaDir从JobTracker端获取到，我们看一下JobTracker端的getStagingAreaDirInternal方法，如下所示：

1 private String getStagingAreaDirInternal(String user) throws IOException {
2 final Path stagingRootDir = newPath(conf.get("mapreduce.jobtracker.staging.root.dir", "/tmp/hadoop/mapred/staging"));
3 final FileSystem fs = stagingRootDir.getFileSystem(conf);
4 return fs.makeQualified(new Path(stagingRootDir, user+"/.staging")).toString();
5 }

最终获取到的StagingArea目录为${mapreduce.jobtracker.staging.root.dir}/${user}/.staging/，例如，如果使用默认的mapreduce.jobtracker.staging.root.dir值，用户为shirdrn，则StagingArea目录/tmp/hadoop/mapred/staging/shirdrn/.staging/。通过Path submitJobDir = new Path(jobStagingArea, jobId.toString());可以得到submitJobDir，假如一个job的ID为job_200912121733_0002，则submitJobDir的值为/tmp/hadoop/mapred/staging/shirdrn/.staging/job_200912121733_0002/

拷贝资源文件

在配置Job的时候，可以指定tmpfiles、tmpjars、tmparchives，JobClient会将对应的资源文件拷贝到指定的目录中，对应目录如下代码所示：

1 Path filesDir = JobSubmissionFiles.getJobDistCacheFiles(submitJobDir);
2 Path archivesDir = JobSubmissionFiles.getJobDistCacheArchives(submitJobDir);
3 Path libjarsDir = JobSubmissionFiles.getJobDistCacheLibjars(submitJobDir);
4 ...
5 Path submitJarFile = JobSubmissionFiles.getJobJar(submitJobDir);
6 job.setJar(submitJarFile.toString());
7 fs.copyFromLocalFile(originalJarFile, submitJarFile);

上面已经知道Job提交目录，可以分别得到对应的资源所在目录：

  • tmpfiles目录：/tmp/hadoop/mapred/staging/shirdrn/.staging/job_200912121733_0002/files
  • tmpjars目录：/tmp/hadoop/mapred/staging/shirdrn/.staging/job_200912121733_0002/libjars
  • tmparchives目录：/tmp/hadoop/mapred/staging/shirdrn/.staging/job_200912121733_0002/archives
  • Job Jar文件：/tmp/hadoop/mapred/staging/shirdrn/.staging/job_200912121733_0002/job.jar

然后，就可以将对应的资源文件拷贝到对应的目录中。

计算并存储Split数据

根据Job配置中设置的InputFormat，计算该Job的数据数据文件是如何进行分片的，代码如下所示：

1 Configuration conf = job.getConfiguration();
2 InputFormat<?, ?> input = ReflectionUtils.newInstance(job.getInputFormatClass(), conf);
3 List<InputSplit> splits = input.getSplits(job);

实际上就是调用InputFormat的getSplits方法，如果不适用Hadoop自带的FileInputFormat的默认getSplits方法实现，可以自定义实现，重写该默认实现逻辑来定义数据数据文件分片的规则。
计算出输入文件的分片信息，然后需要将这些分片数据写入到HDFS供JobTracker查询初始化MapTask，写入分片数据的实现代码：

1 T[] array = (T[]) splits.toArray(new InputSplit[splits.size()]);
2 // sort the splits into order based on size, so that the biggest
3 // go first
4 Arrays.sort(array, new SplitComparator()); // 根据InputSplit的长度做了一个逆序排序
5 JobSplitWriter.createSplitFiles(jobSubmitDir, conf, jobSubmitDir.getFileSystem(conf), array); // 将split及其元数据信息写入HDFS

接着调用JobSplitWriter.createSplitFiles方法存储Split信息，并创建元数据信息，并保存元数据信息。存储Split信息，代码实现如下所示：

01 SerializationFactory factory = new SerializationFactory(conf);
02 int i = 0;
03 long offset = out.getPos();
04 for(T split: array) {
05 long prevCount = out.getPos();
06 Text.writeString(out, split.getClass().getName());
07 Serializer<T> serializer = factory.getSerializer((Class<T>) split.getClass());
08 serializer.open(out);
09 serializer.serialize(split); // 将split序列化写入到HDFS文件中
10 long currCount = out.getPos();
11 String[] locations = split.getLocations();
12 final int max_loc = conf.getInt(MAX_SPLIT_LOCATIONS, 10);
13 if (locations.length > max_loc) {
14 LOG.warn("Max block location exceeded for split: "+ split + " splitsize: " + locations.length + " maxsize: " + max_loc);
15 locations = Arrays.copyOf(locations, max_loc);
16 }
17 info[i++] = new JobSplit.SplitMetaInfo(locations, offset, split.getLength()); // 创建SplitMetaInfo实例
18 offset += currCount - prevCount;
19 }

我们先看一下FileSplit包含的分片内容，如下所示：

1 private Path file;
2 private long start;
3 private long length;
4 private String[] hosts;

在序列化保存FileSplit到HDFS，可以通过查看FileSplit的write方法，如下所示：

1 @Override
2 public void write(DataOutput out) throws IOException {
3 Text.writeString(out, file.toString());
4 out.writeLong(start);
5 out.writeLong(length);
6 }

需要注意的是，这里面并没有将FileSplit的hosts信息保存，而是存储到了SplitMetaInfo中new JobSplit.SplitMetaInfo(locations, offset, split.getLength())。
下面是保存SplitMetaInfo信息的实现：

01 private static void writeJobSplitMetaInfo(FileSystem fs, Path filename,
02 FsPermission p, int splitMetaInfoVersion,
03 JobSplit.SplitMetaInfo[] allSplitMetaInfo) throws IOException {
04 // write the splits meta-info to a file for the job tracker
05 FSDataOutputStream out = FileSystem.create(fs, filename, p);
06 out.write(JobSplit.META_SPLIT_FILE_HEADER); // 写入META头信息：META_SPLIT_FILE_HEADER = "META-SPL".getBytes("UTF-8");
07 WritableUtils.writeVInt(out, splitMetaInfoVersion); // META版本信息：1
08 WritableUtils.writeVInt(out, allSplitMetaInfo.length); // META对象的数量：每个InputSplit对应一个SplitMetaInfo
09 for (JobSplit.SplitMetaInfo splitMetaInfo : allSplitMetaInfo) {
10 splitMetaInfo.write(out); // 每个都进行存储
11 }
12 out.close();
13 }

看一下SplitMetaInfo存储时包含的数据信息：

1 public void write(DataOutput out) throws IOException {
2 WritableUtils.writeVInt(out, locations.length); // location个数
3 for (int i = 0; i < locations.length; i++) {
4 Text.writeString(out, locations[i]); // 写入每一个location位置信息
5 }
6 WritableUtils.writeVLong(out, startOffset); // 偏移量
7 WritableUtils.writeVLong(out, inputDataLength); // 数据长度
8 }

最后，我们看一下这些数据保存的目录和文件情况。前面已经知道Job提交目录，下面看split存储的文件是如何构建的：

1 FSDataOutputStream out = createFile(fs, JobSubmissionFiles.getJobSplitFile(jobSubmitDir), conf);
2 SplitMetaInfo[] info = writeNewSplits(conf, splits, out);

那么split保存的文件为：/tmp/hadoop/mapred/staging/shirdrn/.staging/job_200912121733_0002/job.split。
同样，split元数据信息文件构建如下所示：

1 writeJobSplitMetaInfo(fs,JobSubmissionFiles.getJobSplitMetaFile(jobSubmitDir),
2 new FsPermission(JobSubmissionFiles.JOB_FILE_PERMISSION), splitVersion, info);

split元数据信息文件为：/tmp/hadoop/mapred/staging/shirdrn/.staging/job_200912121733_0002/job.splitmetainfo。

保存Job配置数据

在提交Job到JobTracker之前，还需要保存Job的配置信息，这些配置数据根据用户在MR程序中配置，覆盖默认的配置值，最后保存到XML文件（job.xml）到HDFS，供JobTracker查询。如下代码，创建submitJobFile文件并写入job配置数据：

01 ...
02 Path submitJobFile = JobSubmissionFiles.getJobConfPath(submitJobDir);
03 jobCopy.set("mapreduce.job.dir", submitJobDir.toString());
04 ...
05 // Write job file to JobTracker's fs
06 FSDataOutputStream out = FileSystem.create(fs, submitJobFile, newFsPermission(JobSubmissionFiles.JOB_FILE_PERMISSION));
07 ...
08 try {
09 jobCopy.writeXml(out);
10 } finally {
11 out.close();
12 }

前面已经知道Job提交目录，我们很容易就能得到job.xml文件的存储路径：/tmp/hadoop/mapred/staging/shirdrn/.staging/job_200912121733_0002/job.xml。

最后，所有的数据都已经准备完成，JobClient就可以基于JobSubmissionProtocol协议方法submitJob，提交Job到JobTracker运行。

时间： 2024-04-14 20:45:45

MapReduce V1：Job提交流程之JobClient端分析的相关文章

MapReduce V1：Job提交流程之JobTracker端分析

我们基于Hadoop 1.2.1源码分析MapReduce V1的处理流程.MapReduce V1实现中,主要存在3个主要的分布式进程(角色):JobClient.JobTracker和TaskTracker,我们主要是以这三个角色的实际处理活动为主线,并结合源码,分析实际处理流程. 上一篇我们分析了Job提交过程中JobClient端的处理流程(详见文章 MapReduce V1:Job提交流程之JobClient端分析),这里我们继续详细分析Job提交在JobTracker端的具体流程.通

MapReduce V1：TaskTracker端启动Task流程分析

我们基于Hadoop 1.2.1源码分析MapReduce V1的处理流程. TaskTracker周期性地向JobTracker发送心跳报告,在RPC调用返回结果后,解析结果得到JobTracker下发的运行Task的指令,即LaunchTaskAction,就会在TaskTracker节点上准备运行这个Task.Task的运行是在一个与TaskTracker进程隔离的JVM实例中执行,该JVM实例是通过org.apache.hadoop.mapred.Child来创建的,所以在创建Child

MapReduce V1：JobTracker端Job/Task数据结构

我们基于Hadoop 1.2.1源码分析MapReduce V1的处理流程.在MapReduce程序运行的过程中,JobTracker端会在内存中维护一些与Job/Task运行相关的信息,了解这些内容对分析MapReduce程序执行流程的源码会非常有帮助. 在编写MapReduce程序时,我们是以Job为单位进行编程处理,一个应用程序可能由一组Job组成,而MapReduce框架给我们暴露的只是一些Map和Reduce的函数接口,在运行期它会构建对应MapTask和ReduceTask,所以我们

MapReduce V1：TaskTracker设计要点概要分析

我们基于Hadoop 1.2.1源码分析MapReduce V1的处理流程. 本文不打算深入地详细分析TaskTracker某个具体的处理流程,而是概要地分析TaskTracker在MapReduce框架中的主要负责处理那些事情,是我们能够在宏观上了解TaskTracker端都做了哪些工作.我尽量将TaskTracker端的全部要点内容提出来,但是涉及到详细的分析,只是点到为止,后续会对相应模块的处理流程结合代码进行分析. TaskTracker主要负责MapReduce计算集群中Task运行的

MapReduce V1：JobTracker处理Heartbeat流程分析

我们基于Hadoop 1.2.1源码分析MapReduce V1的处理流程.这篇文章的内容,更多地主要是描述处理/交互流程性的东西,大部分流程图都是经过我梳理后画出来的(开始我打算使用序列图来描述流程,但是发现很多流程在单个对象内部都已经非常复杂,想要通过序列图表达有点担心描述不清,所以选择最基本的程序流程图),可能看起来比较枯燥,重点还是关注主要的处理流程要点,特别的地方我会刻意标示出来,便于理解. JobTracker与TaskTracker之间通过org.apache.hadoop.map

MapReduce V1：MapTask执行流程分析

我们基于Hadoop 1.2.1源码分析MapReduce V1的处理流程. 在文章<MapReduce V1:TaskTracker设计要点概要分析>中我们已经了解了org.apache.hadoop.mapred.Child启动的基本流程,在Child VM启动的过程中会运行MapTask,实际是运行用户编写的MapReduce程序中的map方法中的处理逻辑,我们首先看一下,在Child类中,Child基于TaskUmbilicalProtocol协议与TaskTracker通信,获取到该

Android提权漏洞CVE-2014-7920&amp;CVE-2014-7921分析

 Android提权漏洞CVE-2014-7920&CVE-2014-7921分析 没羽@阿里聚安全 这是Android mediaserver的提权漏洞,利用CVE-2014-7920和CVE-2014-7921实现提权,从0权限提到media权限,其中CVE-2014-7921影响Android 4.0.3及以后的版本,CVE-2014-7920影响Android 2.2及以后的版本.Google直到Android5.1才修复这2个漏洞.该漏洞[1]披露过程如下: 2016年1月24日漏洞作

Zend Framework教程之Zend_Config_Ini用法分析_php实例

本文实例讲述了Zend Framework教程之Zend_Config_Ini用法.分享给大家供大家参考,具体如下: Zend_Config_Ini允许开发者通过嵌套的对象属性语法在应用程序中用熟悉的 INI 格式存储和读取配置数据.INI 格式在提供拥有配置数据键的等级结构和配置数据节之间的继承能力方面具有专长.配置数据等级结构通过用点或者句号 (.)分离键值.一个节可以扩展或者通过在节的名称之后带一个冒号(:)和被继承的配置数据的节的名称来从另一个节继承. parse_ini_file Ze

Hadoop MapReduce最常见的两种容错场景分析

本文将分析Hadoop MapReduce(包括MRv1和MRv2)的两种常见的容错场景,第一种是,作业的某个任务阻塞了,长时间占用资源不释放,如何处理?另外一种是,作 业的Map http://www.aliyun.com/zixun/aggregation/17034.html">Task全部运行完成后,在Reduce Task运行过程中,某个Map Task所在节点挂了,或者某个Map Task结果存放磁盘损坏了,该如何处理? 第一种场景:作业的某个任务阻塞了,长时间占用资源不释放,