MapReduce源码分析之JobSubmitter（一）-低调大师

MapReduce源码分析之JobSubmitter（一）

2016-04-21 519

JobSubmitter，顾名思义，它是MapReduce中作业提交者，而实际上JobSubmitter除了构造方法外，对外提供的唯一一个非private成员变量或方法就是submitJobInternal()方法，它是提交Job的内部方法，实现了提交Job的所有业务逻辑。本文，我们将深入研究MapReduce中用于提交Job的组件JobSubmitter。

首先，我们先看下JobSubmitter的类成员变量，如下：

  // 文件系统FileSystem实例
  private FileSystem jtFs;
  // 客户端通信协议ClientProtocol实例
  private ClientProtocol submitClient;
  // 提交作业的主机名
  private String submitHostName;
  // 提交作业的主机地址
  private String submitHostAddress;

它一共有四个类成员变量，分别为：

1、文件系统FileSystem实例jtFs：用于操作作业运行需要的各种文件等；

2、客户端通信协议ClientProtocol实例submitClient：用于与集群交互，完成作业提交、作业状态查询等；

3、提交作业的主机名submitHostName；

4、提交作业的主机地址submitHostAddress。

其中，客户端通信协议ClientProtocol实例submitClient是通过Cluster的客户端通信协议ClientProtocol实例client来赋值的，我们在《MapReduce源码分析之新API作业提交（二）：连接集群》一文中曾经提到过，它根据MapReduce中参数mapreduce.framework.name的配置为yarn或local，有Yarn模式的YARNRunner和Local模式的LocalJobRunner两种情况。
接下来，我们再看下JobSubmitter的构造函数，如下：

  JobSubmitter(FileSystem submitFs, ClientProtocol submitClient) 
  throws IOException {
	  
	// 根据入参赋值成员变量submitClient、jtFs
    this.submitClient = submitClient;
    this.jtFs = submitFs;
  }

很简单，根据入参赋值成员变量submitClient、jtFs而已。

关键的来了，我们看下JobSubmitter唯一的对外核心功能方法submitJobInternal()，它被用于提交作业至集群，代码如下：

  /**
   * Internal method for submitting jobs to the system.
   * 
   * <p>The job submission process involves:
   * <ol>
   *   <li>
   *   Checking the input and output specifications of the job.
   *   </li>
   *   <li>
   *   Computing the {@link InputSplit}s for the job.
   *   </li>
   *   <li>
   *   Setup the requisite accounting information for the 
   *   {@link DistributedCache} of the job, if necessary.
   *   </li>
   *   <li>
   *   Copying the job's jar and configuration to the map-reduce system
   *   directory on the distributed file-system. 
   *   </li>
   *   <li>
   *   Submitting the job to the <code>JobTracker</code> and optionally
   *   monitoring it's status.
   *   </li>
   * </ol></p>
   * @param job the configuration to submit
   * @param cluster the handle to the Cluster
   * @throws ClassNotFoundException
   * @throws InterruptedException
   * @throws IOException
   */
  JobStatus submitJobInternal(Job job, Cluster cluster) 
  throws ClassNotFoundException, InterruptedException, IOException {

    //validate the jobs output specs 
	// 调用checkSpecs()方法，校验作业输出路径是否配置，且是否已存在，
	// 正确的情况应该是已配置且未存在，输出路径配置参数为mapreduce.output.fileoutputformat.outputdir，
	// 之前WordCount作业的输出路径配置为hdfs://nameservice1/output/output
    checkSpecs(job);

    // 从作业job中获取配置信息conf
    Configuration conf = job.getConfiguration();
    
    // 调用addMRFrameworkToDistributedCache()方法添加应用框架路径到分布式缓存中
    addMRFrameworkToDistributedCache(conf);

    // 通过JobSubmissionFiles的getStagingDir()静态方法获取作业执行时阶段区域路径jobStagingArea
    // 取参数yarn.app.mapreduce.am.staging-dir，参数未配置默认为/tmp/hadoop-yarn/staging
	// 然后后面是/提交作业用户名/.staging
	// 通过之前的WordCount任务的执行，我们查看历史记录，得知参数yarn.app.mapreduce.am.staging-dir配置的为/user，
	// 而提交作业用户名为hdfs，所以完整的路径应该为/user/hdfs/.staging
    Path jobStagingArea = JobSubmissionFiles.getStagingDir(cluster, conf);
    
    //configure the command line options correctly on the submitting dfs
    // 获取当前本机地址
    InetAddress ip = InetAddress.getLocalHost();
    
    // 确定提交作业的主机地址、主机名，并设置入配置信息conf，对应参数分别为
    // mapreduce.job.submithostname
    // mapreduce.job.submithostaddress
    if (ip != null) {
      submitHostAddress = ip.getHostAddress();
      submitHostName = ip.getHostName();
      conf.set(MRJobConfig.JOB_SUBMITHOST,submitHostName);
      conf.set(MRJobConfig.JOB_SUBMITHOSTADDR,submitHostAddress);
    }
    
    // 生成作业ID，即JobID实例jobId
    JobID jobId = submitClient.getNewJobID();
    
    // 将jobId设置入job
    job.setJobID(jobId);
    
    // 构造提交作业路径Path实例submitJobDir，jobStagingArea后接/jobId，比如/job_1459913635503_0005
    // 之前WordCount作业的完整路径为/user/hdfs/.staging/job_1459913635503_0005
    Path submitJobDir = new Path(jobStagingArea, jobId.toString());
    JobStatus status = null;
    
    // 设置作业一些参数：
    try {
    	
      // 设置mapreduce.job.user.name为当前用户，之前的WordCount示例配置的为hdfs用户
      conf.set(MRJobConfig.USER_NAME,
          UserGroupInformation.getCurrentUser().getShortUserName());
      
      // 设置hadoop.http.filter.initializers为AmFilterInitializer
      conf.set("hadoop.http.filter.initializers", 
          "org.apache.hadoop.yarn.server.webproxy.amfilter.AmFilterInitializer");
      
      // 设置mapreduce.job.dir为submitJobDir，比如/user/hdfs/.staging/job_1459913635503_0005
      conf.set(MRJobConfig.MAPREDUCE_JOB_DIR, submitJobDir.toString());
      LOG.debug("Configuring job " + jobId + " with " + submitJobDir 
          + " as the submit dir");
      
      // get delegation token for the dir
      // 获取路径的授权令牌：调用TokenCache的obtainTokensForNamenodes()静态方法
      TokenCache.obtainTokensForNamenodes(job.getCredentials(),
          new Path[] { submitJobDir }, conf);
      
      // 获取密钥和令牌，并将它们存储到令牌缓存TokenCache中
      populateTokenCache(conf, job.getCredentials());

      // generate a secret to authenticate shuffle transfers
      if (TokenCache.getShuffleSecretKey(job.getCredentials()) == null) {
        KeyGenerator keyGen;
        try {
         
          int keyLen = CryptoUtils.isShuffleEncrypted(conf) 
              ? conf.getInt(MRJobConfig.MR_ENCRYPTED_INTERMEDIATE_DATA_KEY_SIZE_BITS, 
                  MRJobConfig.DEFAULT_MR_ENCRYPTED_INTERMEDIATE_DATA_KEY_SIZE_BITS)
              : SHUFFLE_KEY_LENGTH;
          keyGen = KeyGenerator.getInstance(SHUFFLE_KEYGEN_ALGORITHM);
          keyGen.init(keyLen);
        } catch (NoSuchAlgorithmException e) {
          throw new IOException("Error generating shuffle secret key", e);
        }
        SecretKey shuffleKey = keyGen.generateKey();
        TokenCache.setShuffleSecretKey(shuffleKey.getEncoded(),
            job.getCredentials());
      }

      // 复制并且配置相关文件
      copyAndConfigureFiles(job, submitJobDir);

      // 获取配置文件路径：job.xml
      Path submitJobFile = JobSubmissionFiles.getJobConfPath(submitJobDir);
      
      // Create the splits for the job
      LOG.debug("Creating splits at " + jtFs.makeQualified(submitJobDir));
      
      // 调用writeSplits()方法，写分片数据文件job.split和分片元数据文件job.splitmetainfo，
      // 并获得计算得到的map任务数目maps
      int maps = writeSplits(job, submitJobDir);
      
      // 配置信息中设置map任务数目mapreduce.job.maps为上面得到的maps
      conf.setInt(MRJobConfig.NUM_MAPS, maps);
      
      LOG.info("number of splits:" + maps);

      // write "queue admins of the queue to which job is being submitted"
      // to job file.
      
      // 获取作业队列名queue，取参数mapreduce.job.queuename，参数未配置默认为default，
      // 之前的WordCount任务示例中，作业队列名queue就为default
      String queue = conf.get(MRJobConfig.QUEUE_NAME,
          JobConf.DEFAULT_QUEUE_NAME);
      
      // 获取队列的访问权限控制列表AccessControlList实例acl，通过客户端通信协议ClientProtocol实例submitClient的getQueueAdmins()方法，传入队列名queue，
      // 实际上之前的WordCount任务示例中，这里获取的是*
      AccessControlList acl = submitClient.getQueueAdmins(queue);
      
      // 配置信息中设置队列参数mapred.queue.default.acl-administer-jobs
      // 之前的WordCount任务示例中，该参数被设置成为*
      conf.set(toFullPropertyName(queue,
          QueueACL.ADMINISTER_JOBS.getAclName()), acl.getAclString());

      // removing jobtoken referrals before copying the jobconf to HDFS
      // as the tasks don't need this setting, actually they may break
      // because of it if present as the referral will point to a
      // different job.
      // 清空缓存的令牌
      TokenCache.cleanUpTokenReferral(conf);

      // 根据参数确定是否需要追踪令牌ID
      // 取参数mapreduce.job.token.tracking.ids.enabled，参数未配置默认为false
      if (conf.getBoolean(
          MRJobConfig.JOB_TOKEN_TRACKING_IDS_ENABLED,
          MRJobConfig.DEFAULT_JOB_TOKEN_TRACKING_IDS_ENABLED)) {
        
    	// 通过job获取令牌ID，并存储到trackingIds列表中
    	// Add HDFS tracking ids
        ArrayList<String> trackingIds = new ArrayList<String>();
        for (Token<? extends TokenIdentifier> t :
            job.getCredentials().getAllTokens()) {
          trackingIds.add(t.decodeIdentifier().getTrackingId());
        }
        
        // 将trackingIds列表中的内容设置到参数mapreduce.job.token.tracking.ids中
        conf.setStrings(MRJobConfig.JOB_TOKEN_TRACKING_IDS,
            trackingIds.toArray(new String[trackingIds.size()]));
      }

      // Set reservation info if it exists
      // 如有必要，设置存在的预订信息
      // 参数为mapreduce.job.reservation.id
      ReservationId reservationId = job.getReservationId();
      if (reservationId != null) {
        conf.set(MRJobConfig.RESERVATION_ID, reservationId.toString());
      }

      // Write job file to submit dir
      // 调用writeConf()方法，写入作业配置信息至文件job.xml
      writeConf(conf, submitJobFile);
      
      //
      // Now, actually submit the job (using the submit name)
      // 调用printTokens()方法打印令牌信息到Log文件
      printTokens(jobId, job.getCredentials());
      
      // 通过客户端通信协议ClientProtocol实例submitClient的submitJob()方法提交作业，
      // 并获取作业状态JobStatus实例status
      // 由集群连接一文的分析我们可以知道，这个submitClient实际上是YARNRunner或LocalJobRunner对象，
      // 最终调用的是二者的submitJob()方法，我们留待以后分析
      status = submitClient.submitJob(
          jobId, submitJobDir.toString(), job.getCredentials());
      
      // 如果作业状态JobStatus实例status不为null，直接返回，否则抛出无法加载作业的IO异常
      if (status != null) {
        return status;
      } else {
        throw new IOException("Could not launch job");
      }
    } finally {
    	
      // 最终，抛出无法加载作业的IO异常前，调用文件系统FileSystem实例jtFs的delete()方法，
      // 删除作业提交的相关目录或文件submitJobDir
      if (status == null) {
        LOG.info("Cleaning up the staging area " + submitJobDir);
        if (jtFs != null && submitJobDir != null)
          jtFs.delete(submitJobDir, true);

      }
    }
  }

submitJobInternal()方法篇幅比较长，逻辑也很复杂，本文先介绍下它的大体逻辑，后续分文会介绍各个环节的详细内容，且下面涉及到的之前WordCount作业示例在《Hadoop2.6.0版本MapReudce示例之WordCount（一）》及其姊妹篇中，敬请注意！submitJobInternal()方法大体逻辑如下：

1、调用checkSpecs()方法，校验作业输出路径是否配置，且是否已存在：

正确的情况应该是已配置且未存在，输出路径配置参数为mapreduce.output.fileoutputformat.outputdir，之前WordCount作业的输出路径配置为hdfs://nameservice1/output/output；

2、从作业job中获取配置信息conf；

3、调用addMRFrameworkToDistributedCache()方法添加应用框架路径到分布式缓存中；

4、通过JobSubmissionFiles的getStagingDir()静态方法获取作业执行时阶段区域路径jobStagingArea：

取参数yarn.app.mapreduce.am.staging-dir，参数未配置默认为/tmp/hadoop-yarn/staging，然后后面是/提交作业用户名/.staging，通过之前的WordCount任务的执行，我们查看历史记录，得知参数yarn.app.mapreduce.am.staging-dir配置的为/user，而提交作业用户名为hdfs，所以完整的路径应该为/user/hdfs/.staging；

5、获取当前本机地址ip；

6、确定提交作业的主机地址、主机名，并设置入配置信息conf，对应参数分别为mapreduce.job.submithostname、mapreduce.job.submithostaddress；

7、生成作业ID，即JobID实例jobId：

通过客户端通信协议ClientProtocol实例submitClient的getNewJobID()方法生成作业ID，即JobID实例jobId；

8、将jobId设置入job；

9、构造提交作业路径Path实例submitJobDir：

jobStagingArea后接/jobId，比如/job_1459913635503_0005，之前WordCount作业的完整路径为/user/hdfs/.staging/job_1459913635503_0005；

10、设置作业一些参数：

10.1、设置mapreduce.job.user.name为当前用户，之前的WordCount示例配置的为hdfs用户；

10.2、设置hadoop.http.filter.initializers为AmFilterInitializer；

10.3、设置mapreduce.job.dir为submitJobDir，比如/user/hdfs/.staging/job_1459913635503_0005；

11、获取路径的授权令牌：调用TokenCache的obtainTokensForNamenodes()静态方法；

12、通过populateTokenCache()方法获取密钥和令牌，并将它们存储到令牌缓存TokenCache中；

14、复制并且配置相关文件：通过copyAndConfigureFiles()方法实现；

15、获取配置文件路径：job.xml;

16、调用writeSplits()方法，写分片数据文件job.split和分片元数据文件job.splitmetainfo，并获得计算得到的map任务数目maps；

17、配置信息中设置map任务数目mapreduce.job.maps为上面得到的maps；

18、获取作业队列名queue，取参数mapreduce.job.queuename，参数未配置默认为default，之前的WordCount任务示例中，作业队列名queue就为default；

19、获取队列的访问权限控制列表AccessControlList实例acl：

通过客户端通信协议ClientProtocol实例submitClient的getQueueAdmins()方法，传入队列名queue，实际上之前的WordCount任务示例中，这里获取的是*；

20、配置信息中设置队列参数mapred.queue.default.acl-administer-jobs，之前的WordCount任务示例中，该参数被设置成为*；

21、清空缓存的令牌：通过TokenCache的cleanUpTokenReferral()方法实现；

22、根据参数确定是否需要追踪令牌ID，如果需要的话：

取参数mapreduce.job.token.tracking.ids.enabled，参数未配置默认为false，通过job获取令牌ID，并存储到trackingIds列表中，将trackingIds列表中的内容设置到参数mapreduce.job.token.tracking.ids中；

23、如有必要，设置存在的预订信息：参数为mapreduce.job.reservation.id；

24、调用writeConf()方法，写入作业配置信息至文件job.xml；

25、调用printTokens()方法打印令牌信息到Log文件；

26、通过客户端通信协议ClientProtocol实例submitClient的submitJob()方法提交作业，并获取作业状态JobStatus实例status：

由集群连接一文的分析我们可以知道，这个submitClient实际上是YARNRunner或LocalJobRunner对象，最终调用的是二者的submitJob()方法，我们留待以后分析；

27、如果作业状态JobStatus实例status不为null，直接返回，否则抛出无法加载作业的IO异常：

最终，抛出无法加载作业的IO异常前，调用文件系统FileSystem实例jtFs的delete()方法，删除作业提交的相关目录或文件submitJobDir。

整体流程如上，对于关键步骤的主要细节，限于篇幅，敬请关注《MapReduce源码分析之JobSubmitter（二）》！

微信关注我们

原文链接：https://yq.aliyun.com/articles/245618

转载内容版权归作者及来源网站所有！

低调大师中文资讯倾力打造互联网数据资讯、行业资源、电子商务、移动互联网、网络营销平台。持续更新报道IT业界、互联网、市场资讯、驱动更新,是最及时权威的产业资讯及硬件资讯报道平台。

MapReduce源码分析之新API作业提交（二）：连接集群

MapReduce作业提交时连接集群是通过Job的connect()方法实现的，它实际上是构造集群Cluster实例cluster，代码如下： private synchronized void connect() throws IOException, InterruptedException, ClassNotFoundException { // 如果cluster为null，构造Cluster实例cluster， // Cluster为连接MapReduce集群的一种工具，提供了一种获取MapReduce集群信息的方法 if (cluster == null) { cluster = ugi.doAs(new PrivilegedExceptionAction<Cluster>() { public Cluster run() throws IOException, InterruptedException, ClassNotFoundException { return new Cluster(getConfiguration()); } }); } } 这...

2016-04-22

552

MapReduce主要包括两个阶段：一个是Map，一个是Reduce. 每一步都有key-value对作为输入和输出。 Map阶段的key-value对的格式是由输入的格式决定的，如果是默认的TextInputFormat,则每行作为一个记录进程处理，其中key为此行的开头相对文件的起始位置，value就是此行的字符文本。Map阶段的输出的key-value对的格式必须同reduce阶段的输入key-value对的格式相对应。下面开始尝试，假设我们需要处理一批有关天气的数据，其格式如下：按照ASCII码存储，每行一条记录每一行字符从0开始计数，第15个到第18个字符为年第25个到第29个字符为温度，其中第25位是符号+/- Text文本样例： 0067011990999991950051507+0000+ 0043011990999991950051512+0022+ 0043011990999991950051518-0011+ 0043012650999991949032412+0111+ 0043012650999991949032418+0078+ 0067011990999...

2016-04-23

507

资源下载

更多资源

优质分享App

近一个月的开发和优化，本站点的第一个app全新上线。该app采用极致压缩，本体才4.36MB。系统里面做了大量数据访问、缓存优化。方便用户在手机上查看文章。后续会推出HarmonyOS的适配版本。

Mario

马里奥是站在游戏界顶峰的超人气多面角色。马里奥靠吃蘑菇成长，特征是大鼻子、头戴帽子、身穿背带裤，还留着胡子。与他的双胞胎兄弟路易基一起，长年担任任天堂的招牌角色。

腾讯云软件源

为解决软件依赖安装时官方源访问速度慢的问题，腾讯云为一些软件搭建了缓存服务。您可以通过使用腾讯云软件源站来提升依赖包的安装速度。为了方便用户自由搭建服务架构，目前腾讯云软件源站支持公网访问和内网访问。

Sublime Text

Sublime Text具有漂亮的用户界面和强大的功能，例如代码缩略图，Python的插件，代码段等。还可自定义键绑定，菜单和工具栏。Sublime Text 的主要功能包括：拼写检查，书签，完整的 Python API ， Goto 功能，即时项目切换，多选择，多窗口等等。Sublime Text 是一个跨平台的编辑器，同时支持Windows、Linux、Mac OS X等操作系统。