@[toc]

13.MapReduce框架原理

13.1InputFormat数据输入

13.1.1切片与MapTask并行度决定机制

13.1.1.1问题引出

MapTask的并行度决定Map阶段的任务处理并发度，进而影响到整个Job的处理速度  
  思考：1G的数据，启动8个MapTask，可以提高集群的并发处理能力。那么1K的数据，也启动8个MapTask，会提高集群性能吗？MapTask并行任务是否越多越好呢？哪些因素影响了MapTask并行度？

13.1.1.2MapTask并行度决定机制

  数据块：Block是HDFS物理上把数据分成一块一块。数据块是HDFS存储数据单位。
  数据切片：数据切片只是在逻辑上对输入进行分片，并不会在磁盘上将其切分成片进行存储。数据切片是MapReduce程序计算输入数据的单位，一个切片会对应启动一个MapTask。

13.1.2Job提交流程源码

13.1.2.1Job提交流程源码详解

13.1.2.1.1Job提交流程源码断点在哪断并且介绍相关源码：

在WordCountDriver中第七步提交job打上断点

然后在第一个断点处进来后在job类中waitForCompletion中this.submit（）这里打上断点

在this.submit()这里进入submit()方法打上这三处断点

这里是确保你的状态是正确的，如果状态不对，或者状态是RUNNING，会抛异常

在这个setUseNewAPI()方法里进行兼容性处理，因为新老API是不一样的，所以让hadoop1.x和2.x，3.x兼容

this.connect()有两个客户端，一个yarn客户端，一个本地客户端，可以连接yarn客户端或者本地客户端，由于我运行的这个代码是本地模式，所以这里连接的是本地的客户端

从this.connect()这里进来后红色框柱的打上断点

然后提前进入Cluster在初始化这里this.initialize(jobTrackAddr, conf);打上断点

然后到这里进入initialize方法里

这里出现了有yarn客户端

在继续下一步这里变成了本地客户端

这里有两个，一个是yarn，一个是本地客户端，因为代码是本地运行，所以用到的就是LocalClientProtocolProvider

这个方法知道有这两个客户端就够了

这里需要点两下强制进入，因为第一次是进入的job.this，然后退出来在强制进入就进入submitJobInternal方法了

进来之后在这两个地方打上断点

老师的checkSpecs

我的checkSpecs

然后强行进入this.checkSpecs(job);方法里面，在里面打上断点

然后强行进入output.checkOutputSpecs(job);方法，这个是检查路径是否正确，如果输出路径为空，或者输出路径已存在则会抛出异常

在电脑上确实有这个文件

在提交每一个任务的时候都有一个独一无二的jobId

 Path submitJobDir = new Path(jobStagingArea, jobId.toString());

这行代码只是把路径写好了，但是没有去创建带有jobId的路径

在这里打上断点

这块代码快速过掉，这部分是关于缓存相关的处理

这个方法拷贝或者配置一些相关信息，强行进入这个方法

在这里打上断点

看它提交哪些信息，强行进入这个方法

然后再强行进入这个方法

一直下一步，直到过了这个代码，就会把之前加上jobId的路径创建出来，

然后再一直下一步到这里，他要提交一些内容

这里有个jobJar，如果是集群模式的话，当前代码的jar包是一定要上传到集群的，通过客户端方式提交到集群，如果是本地，因为本地就是Local模式，这个jar包就在本地，则不用提交jar包。

所以这块如果是本地模式则不会提交jar包，如果是集群模式，则会提交jar包

这里一定要打断点，writeSplits是切片

运行到这里后，立即会多4个文件

这行代码是设置mapTask的个数，这里就可以看出来切片个数决定mapTask的个数，有几个切片，就开几个mapTask

到这行的时候就会在控制台打印有几个切片

运行完这个会多出两个文件，

job.xml的内容是job运行所需要的参数设置默认值

如果是集群模式，还有jar包，面试官问这类问题是时候问你看过源码吗，回答是：提交三样东西，一个是xml，一个jar包，一个切片信息

没有运行这行代码时，state的值为DEFINE

运行后state的值为RUNNING,说明job开始运行了

这块是监控程序的代码

这行代码是监视并打印job信息

运行完这行代码，之前的文件就都删除了

13.1.2.1.2Job提交流程源码详解

waitForCompletion()

submit();

// 1建立连接
    connect();    
        // 1）创建提交Job的代理
        new Cluster(getConfiguration());
            // （1）判断是本地运行环境还是yarn集群运行环境
            initialize(jobTrackAddr, conf); 

// 2 提交job
submitter.submitJobInternal(Job.this, cluster)

    // 1）创建给集群提交数据的Stag路径
    Path jobStagingArea = JobSubmissionFiles.getStagingDir(cluster, conf);

    // 2）获取jobid ，并创建Job路径
    JobID jobId = submitClient.getNewJobID();

    // 3）拷贝jar包到集群
copyAndConfigureFiles(job, submitJobDir);    
    rUploader.uploadFiles(job, jobSubmitDir);

    // 4）计算切片，生成切片规划文件
writeSplits(job, submitJobDir);
        maps = writeNewSplits(job, jobSubmitDir);
        input.getSplits(job);

    // 5）向Stag路径写XML配置文件
    writeConf(conf, submitJobFile);
    conf.writeXml(out);

    // 6）提交Job,返回提交状态
status = submitClient.submitJob(jobId, submitJobDir.toString(), job.getCredentials());