MapReduce调度与执行原理系列文章

转自:http://blog.csdn.net/jaytalent?viewmode=contents

MapReduce调度与执行原理系列文章

一、MapReduce调度与执行原理之作业提交

二、MapReduce调度与执行原理之作业初始化

三、MapReduce调度与执行原理之任务调度

四、MapReduce调度与执行原理之任务调度(续)

前言:本文旨在理清在Hadoop中一个MapReduce作业(Job)在提交到框架后的整个生命周期过程,权作总结和日后参考,如有问题,请不吝赐教。本文不涉及Hadoop的架构设计,如有兴趣请参考相关书籍和文献。在梳理过程中,我对一些感兴趣的源码也会逐行研究学习,以期强化基础。
作者:Jaytalent
开始日期:2013年9月9日
参考资料:【1】《Hadoop技术内幕--深入解析MapReduce架构设计与实现原理》董西成
                  【2】Hadoop 1.0.0 源码
                            【3】《Hadoop技术内幕--深入解析Hadoop Common和HDFS架构设计与实现原理》蔡斌 陈湘萍
一个MapReduce作业的生命周期大体分为5个阶段【1】:
1. 作业提交与初始化
2. 任务调度与监控
3. 任务运行环境准备
4. 任务执行
5. 作业完成
现逐一学习。
由于作业提交是在客户端完成,而初始化在JobTracker完成,本文只关注前者,后者留待下一篇文章学习研究。
一、作业提交与初始化
以WordCount作业为例,先看作业提交的代码片段:
[java] view plaincopy
 
  1. Job job = new Job(conf, "word count");  
  2. job.setJarByClass(WordCount.class);  
  3. job.setMapperClass(TokenizerMapper.class);  
  4. job.setCombinerClass(IntSumReducer.class);  
  5. job.setReducerClass(IntSumReducer.class);  
  6. job.setOutputKeyClass(Text.class);  
  7. job.setOutputValueClass(IntWritable.class);  
  8. FileInputFormat.addInputPath(job, new Path(otherArgs[0]));  
  9. FileOutputFormat.setOutputPath(job, new Path(otherArgs[1]));  
  10. System.exit(job.waitForCompletion(true) ? 0 : 1);  
这里使用的新的MapReduce API。job.waitForCompletion(true)函数调用开始作业提交过程。接下来,依次调用:job.submit --> JobClient.submitJobInternal方法,真正实现作业提交。在JobClient.submitJobInternal方法中,主要有以下准备工作:
1. 获取作业ID
[java] view plaincopy
 
  1. JobID jobId = jobSubmitClient.getNewJobId();  
作业ID时从JobTracker获取的,这是一次RPC调用,方法为getNewJobId,定义在JobSubmissionProtocol接口。
[java] view plaincopy
 
  1. private JobSubmissionProtocol jobSubmitClient;  
Hadoop的RPC机制是基于动态代理实现的。客户端代码使用RPC类提供的代理对象调用服务器的方法。MapReduce中定义了一系列协议接口用于RPC通信。这些协议包括:
a. JobSubmissionProtocol
b. RefreshUserMappingsProtocol
c. RefreshAuthorizationPolicyProtocol
d. AdminOperationsProtocol
e. InterTrackerProtocol
f. TaskUmbilicalProtocol
前面四个协议用于客户端,最后两个协议位于MapReduce内部。这里使用的getNewJobId方法即协议JobSubmissionProtocol所定义:
[java] view plaincopy
 
  1. /** 
  2.    * Allocate a name for the job. 
  3.    * @return a unique job name for submitting jobs. 
  4.    * @throws IOException 
  5.    */  
  6.   public JobID getNewJobId() throws IOException;  
用户使用该协议通过JobTracker提交作业,查看作业状态等。
2. 作业文件上传
JobClient会根据作业配置信息将作业所需文件上传到JobTracker的文件系统,通常是HDFS。配置信息由JobConf类对象维护。在新的API中,JobConf对象作为JobContext对象的组成部分,作业类Job即继承于JobContext类。
在上传文件前,需要在HDFS上创建必要的目录。上传文件的具体过程从JobClient.submitJobInternal方法中这句调用开始:
[java] view plaincopy
 
  1. copyAndConfigureFiles(jobCopy, submitJobDir);  
在配置了提交副本数(mapred.submit.replication,默认为10)等信息后,主要代码分析如下(为了清晰起见,省略了一些日志和异常处理):
[java] view plaincopy
 
  1. // Retrieve command line arguments placed into the JobConf  
  2. // by GenericOptionsParser.  
  3. String files = job.get("tmpfiles");  
  4. String libjars = job.get("tmpjars");  
  5. String archives = job.get("tmparchives");  
首先,从配置中获取不同类型文件的名称和路径,这些配置在作业提交时从命令行(Hadoop Shell)指定。files表示作业依赖的普通文件,比如文本文件;libjars表示应用程序依赖的第三方jar包;archives表示应用程序使用的多个文件打包而成的压缩文件。
[java] view plaincopy
 
  1. // Create a number of filenames in the JobTracker's fs namespace  
  2. FileSystem fs = submitJobDir.getFileSystem(job);  
  3. submitJobDir = fs.makeQualified(submitJobDir);  
  4. FsPermission mapredSysPerms = new FsPermission(JobSubmissionFiles.JOB_DIR_PERMISSION);  
  5. FileSystem.mkdirs(fs, submitJobDir, mapredSysPerms);  
  6. Path filesDir = JobSubmissionFiles.getJobDistCacheFiles(submitJobDir);  
  7. Path archivesDir = JobSubmissionFiles.getJobDistCacheArchives(submitJobDir);  
  8. Path libjarsDir = JobSubmissionFiles.getJobDistCacheLibjars(submitJobDir);  
接下来,在JobTracker的文件系统(通常为HDFS)的命名空间创建一系列文件路径名,其中包括前述三种文件类型。
有了路径名后,在HDFS上创建路径并将这些文件拷贝到对应的目录中,代码如下:
[java] view plaincopy
 
  1. // add all the command line files/ jars and archive  
  2. // first copy them to jobtrackers filesystem   
  3.   
  4. if (files != null) {  
  5.   FileSystem.mkdirs(fs, filesDir, mapredSysPerms);  
  6.   String[] fileArr = files.split(",");  
  7.   for (String tmpFile: fileArr) {  
  8.     URI tmpURI;  
  9.     tmpURI = new URI(tmpFile);  
  10.      
  11.     Path tmp = new Path(tmpURI);  
  12.     Path newPath = copyRemoteFiles(fs,filesDir, tmp, job, replication);  
  13.     URI pathURI = getPathURI(newPath, tmpURI.getFragment());  
  14.     DistributedCache.addCacheFile(pathURI, job);  
  15.     DistributedCache.createSymlink(job);  
  16.   }  
  17. }  
  18. if (libjars != null) {  
  19.   FileSystem.mkdirs(fs, libjarsDir, mapredSysPerms);  
  20.   String[] libjarsArr = libjars.split(",");  
  21.   for (String tmpjars: libjarsArr) {  
  22.     Path tmp = new Path(tmpjars);  
  23.     Path newPath = copyRemoteFiles(fs, libjarsDir, tmp, job, replication);  
  24.     DistributedCache.addArchiveToClassPath  
  25.       (new Path(newPath.toUri().getPath()), job, fs);  
  26.   }  
  27. }  
  28. if (archives != null) {  
  29.  FileSystem.mkdirs(fs, archivesDir, mapredSysPerms);   
  30.  String[] archivesArr = archives.split(",");  
  31.  for (String tmpArchives: archivesArr) {  
  32.    URI tmpURI;  
  33.    tmpURI = new URI(tmpArchives);  
  34.    Path tmp = new Path(tmpURI);  
  35.    Path newPath = copyRemoteFiles(fs, archivesDir, tmp, job, replication);  
  36.    URI pathURI = getPathURI(newPath, tmpURI.getFragment());  
  37.    DistributedCache.addCacheArchive(pathURI, job);  
  38.    DistributedCache.createSymlink(job);  
  39.  }  
注意,MapReduce作业文件的上传和下载是通过DistributedCache工具完成的,它是一个数据分发工具。用户指定的文件会被分发到各个TaskTracker上以运行Task。这里暂不涉及该工具的细节,留待日后讨论。
最后,将作业对应的jar文件拷贝到HDFS中:
[java] view plaincopy
 
  1. String originalJarPath = job.getJar();  
  2. if (originalJarPath != null) {           // copy jar to JobTracker's fs  
  3.   // use jar name if job is not named.   
  4.   if ("".equals(job.getJobName())){  
  5.     job.setJobName(new Path(originalJarPath).getName());  
  6.   }  
  7.   Path submitJarFile = JobSubmissionFiles.getJobJar(submitJobDir);  
  8.   job.setJar(submitJarFile.toString());  
  9.   fs.copyFromLocalFile(new Path(originalJarPath), submitJarFile);  
  10.   fs.setReplication(submitJarFile, replication);  
  11.   fs.setPermission(submitJarFile,   
  12.       new FsPermission(JobSubmissionFiles.JOB_FILE_PERMISSION));  
  13. }  
注意,在每次上传一种类型的文件后,都会将这种文件的路径配置到JobConf对象中,具体的工作由
[java] view plaincopy
 
  1. DistributedCache.addCacheFile(pathURI, job);  
  2. DistributedCache.addArchiveToClassPath(new Path(newPath.toUri().getPath()), job, fs);  
  3. DistributedCache.addCacheArchive(pathURI, job);  
  4. job.setJar(submitJarFile.toString());  
这四行代码完成。顺便提一句,Path类Hadoop文件系统在java.net.URI的基础上抽象了文件系统中的路径【3】。Java的File类和URL类分别抽象了不同的事物,Path可以说将二者统一起来。
3. 生成InputSplit文件
JobClient调用InputFormat的getSplits方法将用户提交的输入文件生成InputSplit相关信息。
[java] view plaincopy
 
  1. // Create the splits for the job  
  2.    FileSystem fs = submitJobDir.getFileSystem(jobCopy);  
  3.    int maps = writeSplits(context, submitJobDir);  
  4.    jobCopy.setNumMapTasks(maps);  
jobCopy是一个JobConf对象。其中,writeSplits方法会实际调用InputSplit.getSplits方法生成splits信息,并将splits原始信息和元信息写入HDFS对应的目录和文件中。有关split的生成过程日后研究,这里不展开了。最后,将作业对应的JobConf对象以XML配置文件形式写入到HDFS中:
[java] view plaincopy
 
  1. // Write job file to JobTracker's fs          
  2. FSDataOutputStream out =   
  3.    FileSystem.create(fs, submitJobFile,  
  4.        new FsPermission(JobSubmissionFiles.JOB_FILE_PERMISSION));  
  5.   
  6. try {  
  7.    jobCopy.writeXml(out);  
  8.  } finally {  
  9.    out.close();  
  10.  }  
至此,作业文件上传才算正式完毕。
接下来,作业将被提交到JobTracker,请关注下篇文章:
原文地址:https://www.cnblogs.com/cxzdy/p/4943553.html