Shuffle流程是什么

本篇内容介绍了“Shuffle流程是什么”的有关知识，在实际案例的操作过程中，不少人都会遇到这样的困境，接下来就让小编带领大家学习一下如何处理这些情况吧！希望大家仔细阅读，能够学有所成！

创新互联坚持“要么做到，要么别承诺”的工作理念，服务领域包括：成都做网站、网站设计、企业官网、英文网站、手机端网站、网站推广等服务，满足客户于互联网时代的云岩网站设计、移动媒体设计的需求，帮助企业找到有效的互联网解决方案。努力成为您成熟可靠的网络建设合作伙伴！

shuffle流程源码解读

1、从WordCountMapper类中的map方法中写出kv后，进入shuffle流程	--context.write(outK,outV);
进入TaskInputOutputContext中的write()方法			--看下就过
进入WrappedMapper.java中的mapContext.write(key, value);方法	//112行
进入TaskInputOutputContextImpl.java 中output.write(key, value);方法 	//89行
最终定位到MapTask的write()方法内，	//726行

2、重点步骤，收集器对象将kv收集到缓冲区，并在收集前将kv的分区号计算出来.
collector.collect(key, value,partitioner.getPartition(key, value, partitions));
第一次进入该方法时，因为没有设置reduce的个数，所以最终返回的永远是0号分区

3、定位到MapTask类中的collect方法并进入    		//1082行
bufferRemaining -= METASIZE;	//计算缓冲区剩余大小，该行代码前面的代码是对kv类型的一个判断
如果bufferRemaining < 0 则开始进行溢写操作，内部是对数据的一些校验和计算

4、定位到startSpill(); --1126行  	//只有当溢写数据大小满足80%时，才会触发该操作
WordCountMapper持续往缓冲区写数据，当达到溢写条件80%时，开始溢写

5、进入到startSpill()方法内部		--MapTask类1590行
spillReady.signal(); //1602行  		--线程通信, 通知溢写线程开始干活
//执行溢写线程（MapTask内部类SpillThread）的run方法
//run方法中调用MapTask$MapOutputBuffer中的sortAndSpill()方法
直接执行下面的排序和溢写方法		--sortAndSpill()方法  	--MapTask的1605行

6、定位到1615行
final SpillRecord spillRec = new SpillRecord(partitions); //根据分区数创建溢写记录对象
--排序按照分区排序，溢写按照分区溢写

final Path filename =mapOutputFile.getSpillFileForWrite(numSpills, size);//获取溢写文件名称
 ///tmp/hadoop-Administrator/mapred/local/localRunner/Administrator/jobcache/job_local1440922619
 _0001/attempt_local1440922619_0001_m_000000_0/output/（spill0.out），这时还没有溢写文件，只有目录

out = rfs.create(filename);		//创建执行改步后，在上述的目录下生成溢写文件spill0.out文件

7、继续向下走，定位到MapTask类的1625行
sorter.sort(MapOutputBuffer.this, mstart, mend, reporter);		//溢写前排序

8、定位到1629行，进入for循环	--按照分区进行溢写

9、分析for循环内代码，看具体溢写过程
	9.1 先有一个writer对象，通过该对象来完成数据溢写
		writer = new Writer(job, partitionOut, keyClass, valClass, codec,
	9.2 判断是否有设置combinerRunner对象
		如果有，则按照设置的combinerRunner业务去处理；
		如果没有，则走默认的溢写规则

10、执行到1667行，即writer.close();方法，本次溢写完毕，此时我们再去看溢写文件spill0.out文件有数据

11、if (totalIndexCacheMemory >= indexCacheMemoryLimit(大小为：1M)) {}	//MapTask类的1685行
// 如果索引数据超过指定的内存大小，也需要溢写到文件中.（该现象一般情况很难发生.）

12、当本次溢写完毕之后，继续回到WordCountMapper类中的map方法内的context.write(outk,outv);方法处

--说明：因为我们使用本地debug模式调试，所以看不到并行的效果，只能是串行效果，因此看到的是当内存内读取满足
80%时，发生溢写操作，其实溢写并未停止，只不过我们看不到，剩余的溢写数据在20%内存进行

13、如上溢写过程，在整个mapTask中会出现N次,具体多少看数据量. 如果map中最后的数据写到缓冲区，但是没有满足
80%溢写条件的情况，最终也需要将缓冲区的数据刷写到磁盘(最后一次溢写)。

最后一次会发生在 MapTask中关闭 NewOutputCollector对象的时候.
即在该行代码处发生    output.close(mapperContext);	--MapTask的805行

14、进入output.close(mapperContext);方法内	--MapTask的732行
定位到collector.flush();方法 // 735行
-->将缓冲区的数据刷写到磁盘-->重新走sortAndSpill()方法（最后一次刷写）

上述流程，每发生一次溢写就会生成一个溢写小文件（溢写文件内的数据是排好序的）
最终所有的数据都写到磁盘中后，在磁盘上就是多个溢写文件， 比如：spill0.out，spill1.out，...spillN.out

15、溢写全部完成之后，就进入归并操作		--MapTask的1527行
mergeParts();方法，进入该方法，定位到MapTask的1844行
filename[0]: /tmp/hadoop-Administrator/mapred/local/localRunner/Administrator/jobcache/job_local
1440922619_0001/attempt_local1440922619_0001_m_000000_0/output/spill0.out

16、继续向下走，定位到MapTask的1880行
Path finalOutputFile = mapOutputFile.getOutputFileForWrite(finalOutFileSize);
   --归并后，最终输出的文件路径
/tmp/hadoop-Administrator/mapred/local/localRunner/Administrator/jobcache/job_local1440922619_00
01/attempt_local1440922619_0001_m_000000_0/output/file.out

17、继续向下走，定位到MapTask的1882行
Path finalIndexFile = mapOutputFile.getOutputIndexFileForWrite(finalIndexFileSize);
   --归并后，最终输出文件的索引文件
/tmp/hadoop-Administrator/mapred/local/localRunner/Administrator/jobcache/job_local1440922619_00
01/attempt_local1440922619_0001_m_000000_0/output/file.out.index

18、创建file.out 文件
	FSDataOutputStream finalOut = rfs.create(finalOutputFile, true, 4096);

19、for (int parts = 0; parts < partitions; parts++) {}	//1925行，按照分区进行归并排序

20、for循环内具体的归并操作	//1950行
	RawKeyValueIterator kvIter = Merger.merge(job, rfs,
                         keyClass, valClass, codec,
                         segmentList, mergeFactor,
                         new Path(mapId.toString()),
                         job.getOutputKeyComparator(), reporter, sortSegments,
                         null, spilledRecordsCounter, sortPhase.phase(),
                         TaskType.MAP);

21、归并后的数据写出到文件
Writer writer = new Writer(job, finalPartitionOut, 
keyClass, valClass, codec,spilledRecordsCounter); //1961行

//归并也可以使用combiner，但是前提条件是设置了combiner，并且溢写次数大于等于3 
if (combinerRunner == null || numSpills < minSpillsForCombine（3）) {
      Merger.writeFile(kvIter, writer, reporter, job);
} else {
      combineCollector.setWriter(writer);
      combinerRunner.combine(kvIter, combineCollector);
}

22、归并完成
writer.close();		//1972行

23、写出索引文件
spillRec.writeToFile(finalIndexFile, job);	//1986行

24、删除所有的溢写文件spill0.out spill1.out ... spill0.out，只保留最终的输出文件。
for(int i = 0; i < numSpills; i++) {
       rfs.delete(filename[i],true);
}

“Shuffle流程是什么”的内容就介绍到这里了，感谢大家的阅读。如果想了解更多行业相关的知识可以关注创新互联网站，小编将为大家输出更多高质量的实用文章！

网站名称：Shuffle流程是什么
分享URL：http://scjbc.cn/article/pcjedo.html