1.MapReduce的数据流
1) Input -> Mapper阶段
这一阶段的主要分工就是将文件切片和把文件转成K,V对
输入源是一个文件,经过InputFormat之后,到了Mapper就成了K,V对,以上一章的流量案例来说,经过InputFormat之后,变成了手机号为key,这一行数据为value的K,V对,所以这里我们可以自定义InputFormat,按照具体的业务来实现将文件切片并且转为K,V对的方式
2) Mapper -> Reducer阶段
这一阶段叫做shuffle(洗牌)阶段,从Mapper出来的数据是无序的K,V对,那到了Reducer阶段,就变成了有序了.所以我们可以自定义排序规则
3) Reducer -> Output阶段
文件数据经过之前的种种处理,已经变成了有序的数据,这一阶段就是将数据写入文件
2.数据切片与MapTask并行度决定机制
问题引出
MapTask的并行度决定Map阶段的任务处理并发度，进而影响到整个Job的处理速度。
思考：1G的数据，启动8个MapTask，可以提高集群的并发处理能力。那么1K的数据，也启动8个MapTask，会提高集群性能吗？MapTask并行任务是否越多越好呢？哪些因素影响了MapTask并行度？
一个300M的数据,按照blocksize=128M进行存储,在datanode上分别是0-128m,128m-256m,256m-300m
1) 红线切片,将数据按照100M进行切片,每个MapTask处理同样大小的100M数据,看似很公平,datanode1的MapTask处理100M数据,剩下的28m传输给datanode2的MapTask,datanode2的MapTask处理28m+本地的72m数据,剩下的56m再传输给datanode3的MapTask.这样就增加了84m的网络传输数据.为了减少网络传输,yarn有一个本地原则,即block保存在哪个节点上,就在哪个节点上启动MapTask
2) 蓝线切片,切片大小128m=blocksize,每一个datanode启动的MapTask处理的数据都是128m,虽然看似处理速度比1)的100m慢一些,但是却节省了紧张的网络传输资源
所以约定俗成的,切片大小=blocksize
3.Job提交流程源码
确定job的输出路径是否有问题
创建jobid和临时文件夹
生成job的配置信息
经过这一步写入后,文件夹里就有了文件
最后提交job
整理一下
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());
4.FileInputFormat切片机制
上一小章的job提交流程里,生成的临时文件夹里面,有一个job.split文件,说明切片是在job提交之前就切好了,是本地的Driver类做的
那么就再追一下切片源码
最小值minsize=1,最大值maxsize是long的最大值
遍历input源的文件夹,对每一个文件进行切片,先判断文件是否可切
计算切片大小
如,minsize=1,blocksize=32G(33554432),maxsize=9223372036854775807
通过计算,返回的是blocksize.也就是返回的中间值
如果想增加切片大小,那么就增加minsize的大小到>blocksize,如果想减小切片大小,那么就减小maxsize大小到<blocksize
bytesRemaining是文件的大小,根据上面计算到的splitSize来进行切片
404行的SPLIT_SLOP的值是1.1,比的时候是按1.1倍来比,而409行切的时候是按1倍来切
假如blocksize=32M,现在要切一个32.001M的文件,为了多出来的1k启动一个MapTask,很浪费不值当~,所以这样做是为了保证,切出来的片,至少是blocksize大小的10%以上,如果不够10%,那就交给最后的MapTask哥们来处理好了
最后将切片放到切片list里
总结一下