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

标签: mapreduce 调度 原理 | 发表时间:2013-09-09 16:14 | 作者:jtz_MPP
出处:http://blog.csdn.net
前言:本文旨在理清在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作业为例,先看作业提交的代码片段:
    Job job = new Job(conf, "word count");
    job.setJarByClass(WordCount.class);
    job.setMapperClass(TokenizerMapper.class);
    job.setCombinerClass(IntSumReducer.class);
    job.setReducerClass(IntSumReducer.class);
    job.setOutputKeyClass(Text.class);
    job.setOutputValueClass(IntWritable.class);
    FileInputFormat.addInputPath(job, new Path(otherArgs[0]));
    FileOutputFormat.setOutputPath(job, new Path(otherArgs[1]));
    System.exit(job.waitForCompletion(true) ? 0 : 1);
这里使用的新的MapReduce API。job.waitForCompletion(true)函数调用开始作业提交过程。接下来,依次调用:job.submit --> JobClient.submitJobInternal方法,真正实现作业提交。在JobClient.submitJobInternal方法中,主要有以下准备工作:
1. 获取作业ID
JobID jobId = jobSubmitClient.getNewJobId();
作业ID时从JobTracker获取的,这是一次RPC调用,方法为getNewJobId,定义在JobSubmissionProtocol接口。
private JobSubmissionProtocol jobSubmitClient;
Hadoop的RPC机制是基于动态代理实现的。客户端代码使用RPC类提供的代理对象调用服务器的方法。MapReduce中定义了一系列协议接口用于RPC通信。这些协议包括:
a. JobSubmissionProtocol
b. RefreshUserMappingsProtocol
c. RefreshAuthorizationPolicyProtocol
d. AdminOperationsProtocol
e. InterTrackerProtocol
f. TaskUmbilicalProtocol
前面四个协议用于客户端,最后两个协议位于MapReduce内部。这里使用的getNewJobId方法即协议JobSubmissionProtocol所定义:
/**
   * Allocate a name for the job.
   * @return a unique job name for submitting jobs.
   * @throws IOException
   */
  public JobID getNewJobId() throws IOException;
用户使用该协议通过JobTracker提交作业,查看作业状态等。
2. 作业文件上传
JobClient会根据作业配置信息将作业所需文件上传到JobTracker的文件系统,通常是HDFS。配置信息由JobConf类对象维护。在新的API中,JobConf对象作为JobContext对象的组成部分,作业类Job即继承于JobContext类。
在上传文件前,需要在HDFS上创建必要的目录。上传文件的具体过程从JobClient.submitJobInternal方法中这句调用开始:
copyAndConfigureFiles(jobCopy, submitJobDir);
在配置了提交副本数(mapred.submit.replication,默认为10)等信息后,主要代码分析如下(为了清晰起见,省略了一些日志和异常处理):
    // Retrieve command line arguments placed into the JobConf
    // by GenericOptionsParser.
    String files = job.get("tmpfiles");
    String libjars = job.get("tmpjars");
    String archives = job.get("tmparchives");
首先,从配置中获取不同类型文件的名称和路径,这些配置在作业提交时从命令行(Hadoop Shell)指定。files表示作业依赖的普通文件,比如文本文件;libjars表示应用程序依赖的第三方jar包;archives表示应用程序使用的多个文件打包而成的压缩文件。
    // Create a number of filenames in the JobTracker's fs namespace
    FileSystem fs = submitJobDir.getFileSystem(job);
    submitJobDir = fs.makeQualified(submitJobDir);
    FsPermission mapredSysPerms = new FsPermission(JobSubmissionFiles.JOB_DIR_PERMISSION);
    FileSystem.mkdirs(fs, submitJobDir, mapredSysPerms);
    Path filesDir = JobSubmissionFiles.getJobDistCacheFiles(submitJobDir);
    Path archivesDir = JobSubmissionFiles.getJobDistCacheArchives(submitJobDir);
    Path libjarsDir = JobSubmissionFiles.getJobDistCacheLibjars(submitJobDir);
接下来,在JobTracker的文件系统(通常为HDFS)的命名空间创建一系列文件路径名,其中包括前述三种文件类型。
有了路径名后,在HDFS上创建路径并将这些文件拷贝到对应的目录中,代码如下:
    // add all the command line files/ jars and archive
    // first copy them to jobtrackers filesystem 
    
    if (files != null) {
      FileSystem.mkdirs(fs, filesDir, mapredSysPerms);
      String[] fileArr = files.split(",");
      for (String tmpFile: fileArr) {
        URI tmpURI;
        tmpURI = new URI(tmpFile);
       
        Path tmp = new Path(tmpURI);
        Path newPath = copyRemoteFiles(fs,filesDir, tmp, job, replication);
        URI pathURI = getPathURI(newPath, tmpURI.getFragment());
        DistributedCache.addCacheFile(pathURI, job);
        DistributedCache.createSymlink(job);
      }
    }
    if (libjars != null) {
      FileSystem.mkdirs(fs, libjarsDir, mapredSysPerms);
      String[] libjarsArr = libjars.split(",");
      for (String tmpjars: libjarsArr) {
        Path tmp = new Path(tmpjars);
        Path newPath = copyRemoteFiles(fs, libjarsDir, tmp, job, replication);
        DistributedCache.addArchiveToClassPath
          (new Path(newPath.toUri().getPath()), job, fs);
      }
    }
    if (archives != null) {
     FileSystem.mkdirs(fs, archivesDir, mapredSysPerms); 
     String[] archivesArr = archives.split(",");
     for (String tmpArchives: archivesArr) {
       URI tmpURI;
       tmpURI = new URI(tmpArchives);
       Path tmp = new Path(tmpURI);
       Path newPath = copyRemoteFiles(fs, archivesDir, tmp, job, replication);
       URI pathURI = getPathURI(newPath, tmpURI.getFragment());
       DistributedCache.addCacheArchive(pathURI, job);
       DistributedCache.createSymlink(job);
     }
注意,MapReduce作业文件的上传和下载是通过DistributedCache工具完成的,它是一个数据分发工具。用户指定的文件会被分发到各个TaskTracker上以运行Task。这里暂不涉及该工具的细节,留待日后讨论。
最后,将作业对应的jar文件拷贝到HDFS中:
    String originalJarPath = job.getJar();
    if (originalJarPath != null) {           // copy jar to JobTracker's fs
      // use jar name if job is not named. 
      if ("".equals(job.getJobName())){
        job.setJobName(new Path(originalJarPath).getName());
      }
      Path submitJarFile = JobSubmissionFiles.getJobJar(submitJobDir);
      job.setJar(submitJarFile.toString());
      fs.copyFromLocalFile(new Path(originalJarPath), submitJarFile);
      fs.setReplication(submitJarFile, replication);
      fs.setPermission(submitJarFile, 
          new FsPermission(JobSubmissionFiles.JOB_FILE_PERMISSION));
    }
注意,在每次上传一种类型的文件后,都会将这种文件的路径配置到JobConf对象中,具体的工作由
DistributedCache.addCacheFile(pathURI, job);
DistributedCache.addArchiveToClassPath(new Path(newPath.toUri().getPath()), job, fs);
DistributedCache.addCacheArchive(pathURI, job);
job.setJar(submitJarFile.toString());
这四行代码完成。顺便提一句,Path类Hadoop文件系统在java.net.URI的基础上抽象了文件系统中的路径【3】。Java的File类和URL类分别抽象了不同的事物,Path可以说将二者统一起来。
3. 生成InputSplit文件
JobClient调用InputFormat的getSplits方法将用户提交的输入文件生成InputSplit相关信息。
// Create the splits for the job
   FileSystem fs = submitJobDir.getFileSystem(jobCopy);
   int maps = writeSplits(context, submitJobDir);
   jobCopy.setNumMapTasks(maps);
jobCopy是一个JobConf对象。其中,writeSplits方法会实际调用InputSplit.getSplits方法生成splits信息,并将splits原始信息和元信息写入HDFS对应的目录和文件中。有关split的生成过程日后研究,这里不展开了。最后,将作业对应的JobConf对象以XML配置文件形式写入到HDFS中:
    // Write job file to JobTracker's fs        
    FSDataOutputStream out = 
       FileSystem.create(fs, submitJobFile,
           new FsPermission(JobSubmissionFiles.JOB_FILE_PERMISSION));

    try {
       jobCopy.writeXml(out);
     } finally {
       out.close();
     }
至此,作业文件上传才算正式完毕。
接下来,作业将被提交到JobTracker,请关注下篇文章:
MapReduce调度与执行原理之作业初始化





作者:jtz_MPP 发表于2013-9-9 16:14:25 原文链接
阅读:63 评论:0 查看评论

相关 [mapreduce 调度 原理] 推荐:

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

- - CSDN博客云计算推荐文章
前言:本文旨在理清在Hadoop中一个MapReduce作业(Job)在提交到框架后的整个生命周期过程,权作总结和日后参考,如有问题,请不吝赐教. 本文不涉及Hadoop的架构设计,如有兴趣请参考相关书籍和文献. 在梳理过程中,我对一些感兴趣的源码也会逐行研究学习,以期强化基础. 作者:Jaytalent.

MapReduce原理

- - C++博客-牵着老婆满街逛
       MapReduce 是由Google公司的Jeffrey Dean 和 Sanjay Ghemawat 开发的一个针对大规模群组中的海量数据处理的分布式编程模型. MapReduce实现了两个功能. Map把一个函数应用于集合中的所有成员,然后返回一个基于这个处理的结果集. 而Reduce是把从两个或更多个Map中,通过多个线程,进程或者独立系统并行执行处理的结果集进行分类和归纳.

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

- - CSDN博客云计算推荐文章
前言:本文旨在理清在Hadoop中一个MapReduce作业(Job)在提交到框架后的整个生命周期过程,权作总结和日后参考,如有问题,请不吝赐教. 本文不涉及Hadoop的架构设计,如有兴趣请参考相关书籍和文献. 在梳理过程中,我对一些感兴趣的源码也会逐行研究学习,以期强化基础. 作者:Jaytalent.

MapReduce工作原理图文详解

- - ITeye博客
                                                                      MapReduce工作原理图文详解. 前段时间我们云计算团队一起学习了hadoop相关的知识,大家都积极地做了、学了很多东西,收获颇丰. 可是开学后,大家都忙各自的事情,云计算方面的动静都不太大.

Mapreduce小结

- MAGI-CASPER/Peter Pan - 博客园-唯有前进值得敬仰
读完mapreduce论文小结一下. 1.MapReduce是一个编程模型,封装了并行计算、容错、数据分布、负载均衡等细节问题. 输入是一个key-value对的集合,中间输出也是key-value对的集合,用户使用两个函数:Map和Reduce. Map函数接受一个输入的key-value对,然后产生一个中间key-value 对的集合.

Hadoop MapReduce技巧

- - 简单文本
我在使用Hadoop编写MapReduce程序时,遇到了一些问题,通过在Google上查询资料,并结合自己对Hadoop的理解,逐一解决了这些问题. Hadoop对MapReduce中Key与Value的类型是有要求的,简单说来,这些类型必须支持Hadoop的序列化. 为了提高序列化的性能,Hadoop还为Java中常见的基本类型提供了相应地支持序列化的类型,如IntWritable,LongWritable,并为String类型提供了Text类型.

MapReduce优化

- - 行业应用 - ITeye博客
相信每个程序员在 编程时都会问自己两个问题“我如何完成这个任务”,以及“怎么能让程序运行得更快”. 同样,MapReduce计算模型的多次优化也是为了更好地解答这两个问题. MapReduce计算模型的优化涉及了方方面面的内容,但是主要集中在两个方面:一是计算性能方面的优化;二是I/O操作方面的优化.

Spark与Mapreduce?

- - 崔永键的博客
我本人是类似Hive平台的系统工程师,我对MapReduce的熟悉程度是一般,它是我的底层框架. 我隔壁组在实验Spark,想将一部分计算迁移到Spark上. 年初的时候,看Spark的评价,几乎一致表示,Spark是小数据集上处理复杂迭代的交互系统,并不擅长大数据集,也没有稳定性. 但是最近的风评已经变化,尤其是14年10月他们完成了Peta sort的实验,这标志着Spark越来越接近替代Hadoop MapReduce了.

《Hadoop技术内幕:深入解析MapReduce架构设计与实现原理》电子版下载

- - 董的博客
Dong | 新浪微博: 西成懂 | 可以转载, 但必须以超链接形式标明文章原始出处和作者信息及 版权声明. 网址: http://dongxicheng.org/mapreduce-nextgen/hadoop-internals-mapreduce/. 本博客的文章集合: http://dongxicheng.org/recommend/.

Google Percolator替代MapReduce

- Hao - Solidot
Google在新一代内容索引系统中放弃了MapReduce,替代者是尚不为人知的分布式数据处理系统Percolator. The Register报道,Percolator是一种增量处理平台,它能持续更新索引系统,无需从头重新处理一遍整个系统. Google的工程师计划在下个月举行的年度USENIX Symposium 会议上公布Percolator相关论文.