剖析Hadoop和Spark的Shuffle过程差异

四、总结

至此整个shuffle过程完成，***总结几点：

1. shuffle过程就是为了对key进行全局聚合
2. 排序操作伴随着整个shuffle过程，所以Hadoop的shuffle是sort-based的

Spark shuffle相对来说更简单，因为不要求全局有序，所以没有那么多排序合并的操作。Spark shuffle分为write和read两个过程。我们先来看shuffle write。

• 一、shuffle write

shuffle write的处理逻辑会放到该ShuffleMapStage的***(因为spark以shuffle发生与否来划分stage，也就是宽依赖)，final RDD的每一条记录都会写到对应的分区缓存区bucket，如下图所示：明：

1. 上图有2个CPU，可以同时运行两个ShuffleMapTask
2. 每个task将写一个buket缓冲区，缓冲区的数量和reduce任务的数量相等
3. 每个buket缓冲区会生成一个对应ShuffleBlockFile
4. ShuffleMapTask 如何决定数据被写到哪个缓冲区呢?这个就是跟partition算法有关系，这个分区算法可以是hash的，也可以是range的
5. 最终产生的ShuffleBlockFile会有多少呢?就是ShuffleMapTask 数量乘以reduce的数量，这个是非常巨大的

那么有没有办法解决生成文件过多的问题呢?有，开启FileConsolidation即可，开启FileConsolidation之后的shuffle过程如下：一核CPU执行先后执行的ShuffleMapTask可以共用一个bucket缓冲区，然后写到同一份ShuffleFile里去，上图所示的ShuffleFile实际上是用多个ShuffleBlock构成，那么，那么每个worker最终生成的文件数量，变成了cpu核数乘以reduce任务的数量，大大缩减了文件量。

• 二、Shuffle read

Shuffle write过程将数据分片写到对应的分片文件，这时候万事具备，只差去拉取对应的数据过来计算了。

那么Shuffle Read发送的时机是什么?是要等所有ShuffleMapTask执行完，再去fetch数据吗?理论上，只要有一个 ShuffleMapTask执行完，就可以开始fetch数据了，实际上，spark必须等到父stage执行完，才能执行子stage，所以，必须等到所有 ShuffleMapTask执行完毕，才去fetch数据。fetch过来的数据，先存入一个Buffer缓冲区，所以这里一次性fetch的FileSegment不能太大，当然如果fetch过来的数据大于每一个阀值，也是会spill到磁盘的。

fetch的过程过来一个buffer的数据，就可以开始聚合了，这里就遇到一个问题，每次fetch部分数据，怎么能实现全局聚合呢?以word count的reduceByKey(《Spark RDD操作之ReduceByKey 》)为例，假设单词hello有十个，但是一次fetch只拉取了2个，那么怎么全局聚合呢?Spark的做法是用HashMap，聚合操作实际上是map.put(key,map.get(key)+1)，将map中的聚合过的数据get出来相加，然后put回去，等到所有数据fetch完，也就完成了全局聚合。

（编辑：开发网_开封站长网）

【声明】本站内容均来自网络，其相关言论仅代表作者个人观点，不代表本站立场。若无意侵犯到您的权利，请及时与联系站长删除相关内容!