大家好，说到做到，昨天分享了Yarn的资源调度的流程，我也画了一个图来说明，今天分享一下Yarn资源调度的源码分析
MR程序什么时候提价给集群呢？也就是什么时候提交给Yarn，让Yarn来调度呢？
就是代码里的这一句话：
点进入看看Job的这个方法，如下图
进入到sumbit方法
进入connect看看做了什么事情
为cluster赋值，那么cluster是什么呢？
看着UserGroupInformation这个了吗？这个就是对吧当前的用户和hadoop的用户是不是一个用户，就在做本地代码链接hadoop集群的时候报过一个用户不对的异常，就是在这里报出来的，有兴趣的可以进去看看，很简单，这里不再赘述，继续，看最重要的
this.initialize(jobTrackAddr, conf);
这一句话
这里遍历拿到clientProtocol是Yarn，因为在项目下有site.xml配置文件，看代码是怎么实现的
由于上传图片出现问题，直接粘贴代码看
public ClientProtocol create(Configuration conf) throws IOException {
String framework = conf.get("mapreduce.framework.name", "local");
if(!"local".equals(framework)) {
return null;
} else {
conf.setInt("mapreduce.job.maps", 1);
return new LocalJobRunner(conf);
}
}
这里写的很清楚，就是从配置文件中获取mapreduce.framework.name，配置文件中配置了这个是yarn,这个cluster就是yarn的RPC的客户端，通过cluster远程调用yarn的server的相关方法。hadoop中大量使用了RPC这也是分布式中必须要使用到的
继续，接下来就要和yarn通讯了，拿到上传jar包的上传路径，这个上次分享的时候说过的，就是
submitter.submitJobInternal(Job.this, Job.this.cluster);
这句话就完成了与yarn通讯
进去，这么一句话，很熟悉
Path jobStagingArea = JobSubmissionFiles.getStagingDir(cluster, conf);
在提交的时候的那个路径stage就是从这里获取的
接下来就是copy到HDFS上面
this.copyAndConfigureFiles(job, submitJobDir);
这就话就是将job的相关信息，拷贝到那个上传的路径下，并且默认写10个副本，为了从本地获取job信息，为了不走网络，执行快，这里就是一个优化点，如果集群比较大，可是将mapreduce.client.submit.file.replication设置的大一点，为了从本地直接读取job的信息，这一点很重要，来来来，划重点，记住
short replication = (short)conf.getInt("mapreduce.client.submit.file.replication", 10);
提交上去之后到HDFS下看一下文件
[songlj@my01 ~]$ hadoop fs -ls /tmp/hadoop-yarn/staging/songlj/.staging/job_1524231875809_0046
-rw-r--r-- 10 songlj supergroup 109 2018-04-20 11:38 /tmp/hadoop-yarn/staging/songlj/.staging/job_1524231875809_0046/job.split
-rw-r--r-- 3 songlj supergroup 18 2018-04-20 11:38 /tmp/hadoop-yarn/staging/songlj/.staging/job_1524231875809_0046/job.splitmetainfo
-rw-r--r-- 3 songlj supergroup 77459 2018-04-20 11:38 /tmp/hadoop-yarn/staging/songlj/.staging/job_1524231875809_0046/job.xml
那么生成几个map是谁控制的呢？确切的说是MR程序，因为下面这句话在MapReduce包下，还没有在Yarn的相关包里，这个大家记住，来来来，划重点了，这个也是大部分大数据的开发人员，忽略的地方，自然而然的认为是ResourceManager控制的，其实不然，虽然知识点很小，但是这就是检验是否看没有看过源码的验金石。
int maps = this.writeSplits(job, submitJobDir);
这个就是根据split(切片)来确定生成map的数量，这个split生成是有讲究的，大致是一个block就是一个split，这是一般情况，还要看你的设置，默认的情况下就是一个block就是一个split,有兴趣可以进去看一下，这里就不在赘述
接下来就是很重要的，就是shuffle，那么根据昨天画的图接下来应该是第5步，将job加入队列，接下来是由NodeManger领取任务，接下来就是有Resource生成MRAppMaster，Mapper和Resource的调度其实是由这个master来协调的，这里就有一个重要概念就是shuffle，盗用一个很经典的图
这个图如果理解透彻了，对理解其他运算框架storm，spark都有很大的帮助，言归正传，解释一下这个图
1，图中的bugger in memory顾名思义就是内存中的一个缓冲区，其默认的100M，这里还是一个优化点，就是如果服务器的内存比较大，这个值可以设置的大一点，能内存存储运行尽量内存，写入disk需要时间，浪费cpu资源，io等；默认阈值是0.8，就是说如果默认的话，就是当memory达到80M的时候就会触发写入disk操作
2，图中的partition sort and spill disk,这个就是磁盘上的文件，先要分区，排序，分组后再写入磁盘，这里disk上也是一个个的小文件，放一个文件写满了，就会往另一个文件中写
3，图中的merge on disk就是合并后写入disk，也是有分区和排序的。从这里就可以看出MR中有好多磁盘的io，分组排序，这样就会占用很大的资源，这也注定了hadoop做事实的分析，流计算的能力就不是很好，没有spark和storm强，但是hadoop对海量的离线数据的分析还是很厉害的，这也是hadoop没有被淘汰的一个很重要的原因，鼻祖能被淘汰吗？我想到了诺基亚，开个玩笑
4，图中绿色的箭头other maps这个说明这张图的意思是有4个map同时执行的，并且每一个都分了三个partions,那么就会生成3个reduces,reduces的数量可以在代码中继承修改的。这个以后分享说明，这里不再赘述，知道就行了
5，问题来了，map执行完了，reduce task怎么知道他要从哪里才能获取到第一个partions的数据呢？这个是由MRAppMaster来调度的，前面也有说明，master会收到map执行完的反馈信息后，将map执行结果以及map的host都会告诉reduce,reduce通过网络将数据拷贝到reduce所在的服务器上
6，图中reduce Task中进行了好多merge合并，这期间也产生了好多磁盘的io，最终output出去，可以使HDFS，也可以是数据库，也可以是本地文件等等
好了，基本上也差不多了
望指正，不吝赐教！
