Job的提交—客户端

﹏ヽ暗。殇╰゛Y 2022-06-17 09:00 248阅读 0赞

使用过Hadoop的MapReduce API的人大概都知道在我们配置好作业之后，就可以向JobTracker提交该作业了，然后JobTracker才能安排适当的TaskTracker来完成该作业。那么MapReduce在这个过程中到底做了那些事情呢？这就是本文以及接下来的一片博文将要讨论的问题，当然本文主要是围绕客户端在作业的提交过程中的工作来展开。先从全局来把握这个过程吧！

![0_1322914304zNN0.gif][]

在Hadoop中，作业是使用Job对象来抽象的，对于Job，我首先不得不介绍它的一个大家伙JobClient——客户端的实际工作者。JobClient除了自己完成一部分必要的工作外，还负责与JobTracker进行交互。所以客户端对Job的提交，绝大部分都是JobClient完成的，从上图中，我们可以得知JobClient提交Job的详细流程主要如下：

![0_1322915142uq8n.gif][]

JobClient在获取了JobTracker为Job分配的id之后，会在JobTracker的系统目录(HDFS)下为该Job创建一个单独的目录，目录的名字即是Job的id，该目录下会包含文件job.xml、job.jar、job.split等，其中，job.xml文件记录了Job的详细配置信息，job.jar保存了用户定义的关于job的map、reduce操纵，job.split保存了job输入数据的切分信息。在上面的流程图中，我想详细阐述的是JobClient是任何配置Job的运行环境，以及如何对Job的输入数据进行切分。

1.  配置Job的运行环境

Job的运行可能需要依赖一些第三方的jar包、archive文件、或者其他的文件，不过没关系这些都可以通过配置文件中的tmpfiles、tmpjars、tmparchies项来设置。所以在配置Job的运行环境过程中，JobClient会根据这些配置项的设置情况在Job的目录下创建对应的目录，并把这些文件从本地copy到这些对应的目录中。

2.  切分Job的输入数据

Job的输入数据的切分策略可以有用户自己来定义(org.apache.hadoop.mapreduce.InputFormat的实现)，但是最后每一个输入数据的切片必须被表示成一个InputSplit对象，同时这些InputSplit对象最后也会保存到Job目录下的job.split文件中。实际上，对输入数据的切分就是对Job的划分，有多少个split，就会有多少个map任务。这里我只要会集体的谈一谈它的一个实现——org.apache.hadoop.mapreduce.lib.input.FileInputFormat：

public List<InputSplit> getSplits(JobContext job) throws IOException {
        long minSize = Math.max(getFormatMinSplitSize(), getMinSplitSize(job));
        long maxSize = getMaxSplitSize(job);
        //对输入文件进行切分
        List<InputSplit> splits = new ArrayList<InputSplit>();
        for (FileStatus file: listStatus(job)) {
          Path path = file.getPath();
          FileSystem fs = path.getFileSystem(job.getConfiguration());
          long length = file.getLen();
    
          //获取该输入文件每一个块的位置信息
    
          BlockLocation[] blkLocations = fs.getFileBlockLocations(file, 0, length);
          if ((length != 0) && isSplitable(job, path)) {
            long blockSize = file.getBlockSize();
            long splitSize = computeSplitSize(blockSize, minSize, maxSize); //计算切片的大小
    
            long bytesRemaining = length;
            while (((double) bytesRemaining)/splitSize > SPLIT_SLOP) {
              int blkIndex = getBlockIndex(blkLocations, length-bytesRemaining);
              splits.add(new FileSplit(path, length-bytesRemaining, splitSize, blkLocations[blkIndex].getHosts()));
              bytesRemaining -= splitSize;
            }
            
            if (bytesRemaining != 0) {
              splits.add(new FileSplit(path, length-bytesRemaining, bytesRemaining,blkLocations[blkLocations.length-1].getHosts()));
            }
          } else if (length != 0) {
            splits.add(new FileSplit(path, 0, length, blkLocations[0].getHosts()));
          } else {
            //Create empty hosts array for zero length files
            splits.add(new FileSplit(path, 0, length, new String[0]));
          }
        }
    
        return splits;
    }

从代码我们可以看到，切片的大小对任务的执行效率至关重要，因此我们因该尽量保证一个切片恰好包含整数个block，或者一个block恰好被整数个split。那么如何来定设置split的大小呢，这是由computeSplitSize方法来确定的：

protected long computeSplitSize(long blockSize, long minSize,long maxSize) {
        return Math.max(minSize, Math.min(maxSize, blockSize));
    }

其中，minSize、maxSize取决于配置文件中的mapred.min.split.size和mapred.max.split.size项。

最后，结合Job被Mapreduce执行的过程给出一张详细的处理流图。

![0_1322920445xXrT.gif][]

[0_1322914304zNN0.gif]: /images/20220617/197190f194d34a2692045300a9f01c38.png
[0_1322915142uq8n.gif]: /images/20220617/87c06eac2ce44807bf4f709585d04fc1.png
[0_1322920445xXrT.gif]: /images/20220617/c47554221bde4008ad19433ce8cdbc46.png