@[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());