经验首页 前端设计 程序设计 Java相关 移动开发 数据库/运维 软件/图像 大数据/云计算 其他经验
当前位置：技术经验 » 大数据/云/AI » Hadoop » 查看文章
MapReduce之Job提交流程源码和切片源码分析
来源：cnblogs　　作者：kocdaniel　　时间：2019/9/30 8:51:24　　对本文有异议

hadoop2.7.2 MapReduce Job提交源码及切片源码分析

  1. 首先从waitForCompletion函数进入
  1. boolean result = job.waitForCompletion(true);
  1. /**
  2.    * Submit the job to the cluster and wait for it to finish.
  3.    * @param verbose print the progress to the user
  4.    * @return true if the job succeeded
  5.    * @throws IOException thrown if the communication with the 
  6.    *         <code>JobTracker</code> is lost
  7.    */
  8.   public boolean waitForCompletion(boolean verbose
  9.                                    ) throws IOException, InterruptedException,
  10.                                             ClassNotFoundException {
  11.     // 首先判断state，当state为DEFINE时可以提交，进入 submit() 方法
  12.     if (state == JobState.DEFINE) {
  13.       submit();
  14.     }
  15.     if (verbose) {
  16.       monitorAndPrintJob();
  17.     } else {
  18.       // get the completion poll interval from the client.
  19.       int completionPollIntervalMillis = 
  20.         Job.getCompletionPollInterval(cluster.getConf());
  21.       while (!isComplete()) {
  22.         try {
  23.           Thread.sleep(completionPollIntervalMillis);
  24.         } catch (InterruptedException ie) {
  25.         }
  26.       }
  27.     }
  28.     return isSuccessful();
  29.   }
  1. 进入submit()方法
  1. /**
  2.    * Submit the job to the cluster and return immediately.
  3.    * @throws IOException
  4.    */
  5.   public void submit() 
  6.          throws IOException, InterruptedException, ClassNotFoundException {
  7.     // 确认JobState状态为可提交状态，否则不能提交
  8.     ensureState(JobState.DEFINE);
  9.     // 设置使用最新的API
  10.     setUseNewAPI();
  11.     // 进入connect()方法，MapReduce作业提交时连接集群是通过Job类的connect()方法实现的，
  12.     // 它实际上是构造集群Cluster实例cluster
  13.     connect();
  14.     // connect()方法执行完之后，定义提交者submitter
  15.     final JobSubmitter submitter = 
  16.         getJobSubmitter(cluster.getFileSystem(), cluster.getClient());
  17.     status = ugi.doAs(new PrivilegedExceptionAction<JobStatus>() {
  18.       public JobStatus run() throws IOException, InterruptedException, 
  19.       ClassNotFoundException {
  20.         // 这里的核心方法是submitJobInternal(),顾名思义，提交job的内部方法，实现了提交job的所有业务逻辑
  21.           // 进入submitJobInternal
  22.         return submitter.submitJobInternal(Job.this, cluster);
  23.       }
  24.     });
  25.     // 提交之后state状态改变
  26.     state = JobState.RUNNING;
  27.     LOG.info("The url to track the job: " + getTrackingURL());
  28.    }
  1. 进入connect()方法
  • MapReduce作业提交时连接集群通过Job的Connect方法实现，它实际上是构造集群Cluster实例cluster
  • cluster是连接MapReduce集群的一种工具，提供了获取MapReduce集群信息的方法
  • 在Cluster内部，有一个与集群进行通信的客户端通信协议ClientProtocol的实例client，它由ClientProtocolProvider的静态create()方法构造
  • 在create内部，Hadoop2.x中提供了两种模式的ClientProtocol，分别为Yarn模式的YARNRunner和Local模式的LocalJobRunner，Cluster实际上是由它们负责与集群进行通信的
  1.   private synchronized void connect()
  2.           throws IOException, InterruptedException, ClassNotFoundException {
  3.     if (cluster == null) {// cluster提供了远程获取MapReduce的方法
  4.       cluster = 
  5.         ugi.doAs(new PrivilegedExceptionAction<Cluster>() {
  6.                    public Cluster run()
  7.                           throws IOException, InterruptedException, 
  8.                                  ClassNotFoundException {
  9.                      // 只需关注这个Cluster()构造器，构造集群cluster实例
  10.                      return new Cluster(getConfiguration());
  11.                    }
  12.                  });
  13.     }
  14.   }
  1. 进入Cluster()构造器
  1. // 首先调用一个参数的构造器，间接调用两个参数的构造器
  2. public Cluster(Configuration conf) throws IOException {
  3.     this(null, conf);
  4.   }
  6.   public Cluster(InetSocketAddress jobTrackAddr, Configuration conf) 
  7.       throws IOException {
  8.     this.conf = conf;
  9.     this.ugi = UserGroupInformation.getCurrentUser();
  10.     // 最重要的initialize方法
  11.     initialize(jobTrackAddr, conf);
  12.   }
  13.   
  14. // cluster中要关注的两个成员变量是客户端通讯协议提供者ClientProtocolProvider和客户端通讯协议ClientProtocol实例client
  15.   private void initialize(InetSocketAddress jobTrackAddr, Configuration conf)
  16.       throws IOException {
  18.     synchronized (frameworkLoader) {
  19.       for (ClientProtocolProvider provider : frameworkLoader) {
  20.         LOG.debug("Trying ClientProtocolProvider : "
  21.             + provider.getClass().getName());
  22.         ClientProtocol clientProtocol = null; 
  23.         try {
  24.           // 如果配置文件没有配置YARN信息，则构建LocalRunner,MR任务本地运行
  25.           // 如果配置文件有配置YARN信息，则构建YarnRunner,MR任务在YARN集群上运行
  26.           if (jobTrackAddr == null) {
  27.             // 客户端通讯协议client是调用ClientProtocolProvider的create()方法实现
  28.             clientProtocol = provider.create(conf);
  29.           } else {
  30.             clientProtocol = provider.create(jobTrackAddr, conf);
  31.           }
  33.           if (clientProtocol != null) {
  34.             clientProtocolProvider = provider;
  35.             client = clientProtocol;
  36.             LOG.debug("Picked " + provider.getClass().getName()
  37.                 + " as the ClientProtocolProvider");
  38.             break;
  39.           }
  40.           else {
  41.             LOG.debug("Cannot pick " + provider.getClass().getName()
  42.                 + " as the ClientProtocolProvider - returned null protocol");
  43.           }
  44.         } 
  45.         catch (Exception e) {
  46.           LOG.info("Failed to use " + provider.getClass().getName()
  47.               + " due to error: ", e);
  48.         }
  49.       }
  50.     }
  52.     if (null == clientProtocolProvider || null == client) {
  53.       throw new IOException(
  54.           "Cannot initialize Cluster. Please check your configuration for "
  55.               + MRConfig.FRAMEWORK_NAME
  56.               + " and the correspond server addresses.");
  57.     }
  58.   }
  1. 进入submitJobInternal()，job的内部提交方法，用于提交job到集群
  1. JobStatus submitJobInternal(Job job, Cluster cluster) 
  2.   throws ClassNotFoundException, InterruptedException, IOException {
  4.     //validate the jobs output specs 
  5.     // 检查结果的输出路径是否已经存在，如果存在会报异常
  6.     checkSpecs(job);
  8.     // conf里边是集群的xml配置文件信息
  9.     Configuration conf = job.getConfiguration();
  10.     // 添加MR框架到分布式缓存中
  11.     addMRFrameworkToDistributedCache(conf);
  13.     // 获取提交执行时相关资源的临时存放路径
  14.     // 参数未配置时默认是（工作空间根目录下的）/tmp/hadoop-yarn/staging/提交作业用户名/.staging
  15.     Path jobStagingArea = JobSubmissionFiles.getStagingDir(cluster, conf);
  16.     //configure the command line options correctly on the submitting dfs
  17.     InetAddress ip = InetAddress.getLocalHost();
  18.     if (ip != null) {//记录提交作业的主机IP、主机名，并且设置配置信息conf
  19.       submitHostAddress = ip.getHostAddress();
  20.       submitHostName = ip.getHostName();
  21.       conf.set(MRJobConfig.JOB_SUBMITHOST,submitHostName);
  22.       conf.set(MRJobConfig.JOB_SUBMITHOSTADDR,submitHostAddress);
  23.     }
  24.     // 获取JobId
  25.     JobID jobId = submitClient.getNewJobID();
  26.     // 设置jobId
  27.     job.setJobID(jobId);
  28.     // 提交作业的路径Path(Path parent, String child)，会将两个参数拼接为一个路径
  29.     Path submitJobDir = new Path(jobStagingArea, jobId.toString());
  30.     // job的状态
  31.     JobStatus status = null;
  32.     try {
  33.       conf.set(MRJobConfig.USER_NAME,
  34.           UserGroupInformation.getCurrentUser().getShortUserName());
  35.       conf.set("hadoop.http.filter.initializers", 
  36.           "org.apache.hadoop.yarn.server.webproxy.amfilter.AmFilterInitializer");
  37.       conf.set(MRJobConfig.MAPREDUCE_JOB_DIR, submitJobDir.toString());
  38.       LOG.debug("Configuring job " + jobId + " with " + submitJobDir 
  39.           + " as the submit dir");
  40.       // get delegation token for the dir
  41.       TokenCache.obtainTokensForNamenodes(job.getCredentials(),
  42.           new Path[] { submitJobDir }, conf);
  43.       
  44.       populateTokenCache(conf, job.getCredentials());
  46.       // generate a secret to authenticate shuffle transfers
  47.       if (TokenCache.getShuffleSecretKey(job.getCredentials()) == null) {
  48.         KeyGenerator keyGen;
  49.         try {
  50.           keyGen = KeyGenerator.getInstance(SHUFFLE_KEYGEN_ALGORITHM);
  51.           keyGen.init(SHUFFLE_KEY_LENGTH);
  52.         } catch (NoSuchAlgorithmException e) {
  53.           throw new IOException("Error generating shuffle secret key", e);
  54.         }
  55.         SecretKey shuffleKey = keyGen.generateKey();
  56.         TokenCache.setShuffleSecretKey(shuffleKey.getEncoded(),
  57.             job.getCredentials());
  58.       }
  59.       if (CryptoUtils.isEncryptedSpillEnabled(conf)) {
  60.         conf.setInt(MRJobConfig.MR_AM_MAX_ATTEMPTS, 1);
  61.         LOG.warn("Max job attempts set to 1 since encrypted intermediate" +
  62.                 "data spill is enabled");
  63.       }
  65.       // 拷贝jar包到集群
  66.       // 此方法中调用如下方法：rUploader.uploadFiles(job, jobSubmitDir);
  67.       // uploadFiles方法将jar包拷贝到集群
  68.       copyAndConfigureFiles(job, submitJobDir);
  70.       Path submitJobFile = JobSubmissionFiles.getJobConfPath(submitJobDir);
  71.       
  72.       // Create the splits for the job
  73.       LOG.debug("Creating splits at " + jtFs.makeQualified(submitJobDir));
  74.       // 计算切片，生成切片规划文件
  75.       int maps = writeSplits(job, submitJobDir);
  76.       conf.setInt(MRJobConfig.NUM_MAPS, maps);
  77.       LOG.info("number of splits:" + maps);
  79.       // write "queue admins of the queue to which job is being submitted"
  80.       // to job file.
  81.       String queue = conf.get(MRJobConfig.QUEUE_NAME,
  82.           JobConf.DEFAULT_QUEUE_NAME);
  83.       AccessControlList acl = submitClient.getQueueAdmins(queue);
  84.       conf.set(toFullPropertyName(queue,
  85.           QueueACL.ADMINISTER_JOBS.getAclName()), acl.getAclString());
  87.       // removing jobtoken referrals before copying the jobconf to HDFS
  88.       // as the tasks don't need this setting, actually they may break
  89.       // because of it if present as the referral will point to a
  90.       // different job.
  91.       TokenCache.cleanUpTokenReferral(conf);
  93.       if (conf.getBoolean(
  94.           MRJobConfig.JOB_TOKEN_TRACKING_IDS_ENABLED,
  95.           MRJobConfig.DEFAULT_JOB_TOKEN_TRACKING_IDS_ENABLED)) {
  96.         // Add HDFS tracking ids
  97.         ArrayList<String> trackingIds = new ArrayList<String>();
  98.         for (Token<? extends TokenIdentifier> t :
  99.             job.getCredentials().getAllTokens()) {
  100.           trackingIds.add(t.decodeIdentifier().getTrackingId());
  101.         }
  102.         conf.setStrings(MRJobConfig.JOB_TOKEN_TRACKING_IDS,
  103.             trackingIds.toArray(new String[trackingIds.size()]));
  104.       }
  106.       // Set reservation info if it exists
  107.       ReservationId reservationId = job.getReservationId();
  108.       if (reservationId != null) {
  109.         conf.set(MRJobConfig.RESERVATION_ID, reservationId.toString());
  110.       }
  112.       // Write job file to submit dir
  113.       writeConf(conf, submitJobFile);
  114.       
  115.       //
  116.       // Now, actually submit the job (using the submit name)
  117.       // 开始正式提交job
  118.       printTokens(jobId, job.getCredentials());
  119.       status = submitClient.submitJob(
  120.           jobId, submitJobDir.toString(), job.getCredentials());
  121.       if (status != null) {
  122.         return status;
  123.       } else {
  124.         throw new IOException("Could not launch job");
  125.       }
  126.     } finally {
  127.       if (status == null) {
  128.         LOG.info("Cleaning up the staging area " + submitJobDir);
  129.         if (jtFs != null && submitJobDir != null)
  130.           jtFs.delete(submitJobDir, true);
  132.       }
  133.     }
  134.   }
  1. 进入writeSplits(job, submitJobDir)，计算切片，生成切片规划文件
  • 内部会调用writeNewSplits(job, jobSubmitDir)方法
  • writeNewSplits(job, jobSubmitDir)内部定义了一个InputFormat类型的实例input
  • InputFormat主要作用：
    • 验证job的输入规范
    • 对输入的文件进行切分，形成多个InputSplit（切片）文件，每一个InputSplit对应着一个map任务（MapTask）
    • 将切片后的数据按照规则形成key，value键值对RecordReader
  • input调用getSplits（）方法：List<InputSplit> splits = input.getSplits(job);
  1. 进入FileInputFormat类下的getSplits(job)方法
  1. /** 
  2.    * Generate the list of files and make them into FileSplits.
  3.    * @param job the job context
  4.    * @throws IOException
  5.    */
  6.   public List<InputSplit> getSplits(JobContext job) throws IOException {
  7.     StopWatch sw = new StopWatch().start();
  8.       
  9.     // getFormatMinSplitSize()返回值固定为1，getMinSplitSize(job)返回job大小
  10.     long minSize = Math.max(getFormatMinSplitSize(), getMinSplitSize(job));
  11.     // getMaxSplitSize(job)返回Lang类型的最大值
  12.     long maxSize = getMaxSplitSize(job);
  14.     // generate splits 生成切片
  15.     List<InputSplit> splits = new ArrayList<InputSplit>();
  16.     List<FileStatus> files = listStatus(job);
  17.     // 遍历job下的所有文件
  18.     for (FileStatus file: files) {
  19.       // 获取文件路径
  20.       Path path = file.getPath();
  21.       // 获取文件大小
  22.       long length = file.getLen();
  23.       if (length != 0) {
  24.         BlockLocation[] blkLocations;
  25.         if (file instanceof LocatedFileStatus) {
  26.           blkLocations = ((LocatedFileStatus) file).getBlockLocations();
  27.         } else {
  28.           FileSystem fs = path.getFileSystem(job.getConfiguration());
  29.           blkLocations = fs.getFileBlockLocations(file, 0, length);
  30.         }
  31.         // 判断是否可分割
  32.         if (isSplitable(job, path)) {
  33.           // 获取块大小
  34.           // 本地环境块大小默认为32MB，YARN环境在hadoop2.x新版本为128MB，旧版本为64MB
  35.           long blockSize = file.getBlockSize();
  36.           // 计算切片的逻辑大小，默认等于块大小
  37.           // 返回值为：return Math.max(minSize, Math.min(maxSize, blockSize));
  38.           // 其中minSize=1， maxSize=Long类型最大值， blockSize为切片大小
  39.           long splitSize = computeSplitSize(blockSize, minSize, maxSize);
  41.           long bytesRemaining = length;
  42.           // 每次切片时就要判断切片剩下的部分是否大于切片大小的SPLIT_SLOP（默认为1.1）倍，
  43.           // 否则就不再切分，划为一块
  44.           while (((double) bytesRemaining)/splitSize > SPLIT_SLOP) {
  45.             int blkIndex = getBlockIndex(blkLocations, length-bytesRemaining);
  46.             splits.add(makeSplit(path, length-bytesRemaining, splitSize,
  47.                         blkLocations[blkIndex].getHosts(),
  48.                         blkLocations[blkIndex].getCachedHosts()));
  49.             bytesRemaining -= splitSize;
  50.           }
  52.           if (bytesRemaining != 0) {
  53.             int blkIndex = getBlockIndex(blkLocations, length-bytesRemaining);
  54.             splits.add(makeSplit(path, length-bytesRemaining, bytesRemaining,
  55.                        blkLocations[blkIndex].getHosts(),
  56.                        blkLocations[blkIndex].getCachedHosts()));
  57.           }
  58.         } else { // not splitable
  59.           splits.add(makeSplit(path, 0, length, blkLocations[0].getHosts(),
  60.                       blkLocations[0].getCachedHosts()));
  61.         }
  62.       } else { 
  63.         //Create empty hosts array for zero length files
  64.         splits.add(makeSplit(path, 0, length, new String[0]));
  65.       }
  66.     }
  67.     // Save the number of input files for metrics/loadgen
  68.     job.getConfiguration().setLong(NUM_INPUT_FILES, files.size());
  69.     sw.stop();
  70.     if (LOG.isDebugEnabled()) {
  71.       LOG.debug("Total # of splits generated by getSplits: " + splits.size()
  72.           + ", TimeTaken: " + sw.now(TimeUnit.MILLISECONDS));
  73.     }
  74.     return splits;
  75.   }

原文链接：http://www.cnblogs.com/kocdaniel/p/11609404.html

 友情链接：直通硅谷 　点职佳 　北美留学生论坛

Hadoop热门文章

  • Hive中的去重 ：distinct,group by与ROW_Number()窗口函数
  • HBASE的安装过程及运行HBASE程序的需要配置的内容
  • hadoop--Unable to load native-hadoop library for your platform解决方法
  • Hive sampling 语法之TABLESAMPLE用法理解
  • idea项目运行时报错提示Could not locate executable null\bin\winutils.exe in the Hadoop binaries
  • hadoop 无法访问50070端口问题解决
  • Hadoop运行时遇到java.io.FileNotFoundException错误的解决方法
  • KUDU数据导入尝试一：TextFile数据导入Hive,Hive数据导入KUDU
  • Hive安装
  • Hadoop 学习笔记 生态

    • Hadoop推荐文章

  • 基于Hive的大数据分析系统
  • HBase 在统一内容平台业务的优化实践
  • hive窗口分析函数使用详解系列一
  • 两台实体机器4个虚拟机节点的Hadoop集群搭建（Ubuntu版）
  • 在阿里云和腾讯云的轻量应用服务器上搭建Hadoop集群
  • Haproxy搭建 Web 群集实现负载均衡
  • HBase Compaction 原理与线上调优实践
  • Hive执行计划之hive依赖及权限查询和常见使用场景
  • 工良出品：包教会，Hadoop、Hive 搭建部署简易教程
  • Centos7系统编译Hadoop3.3.4

    • 本站QQ群：前端 618073944 | Java 606181507 | Python 626812652 | C/C++ 612253063 | 微信 634508462 | 苹果 692586424 | C#/.net 182808419 | PHP 305140648 | 运维 608723728

    W3xue 的所有内容仅供测试，对任何法律问题及风险不承担任何责任。通过使用本站内容随之而来的风险与本站无关。
    关于我们  |  意见建议  |  捐助我们  |  报错有奖  |  广告合作、友情链接（目前9元/月）请联系QQ：27243702 沸活量
    皖ICP备17017327号-2　皖公网安备34020702000426号