Shuffle原理及对应的Consolidation优化机制是怎样的


这篇文章给大家介绍Shuffle原理及对应的Consol 香港云主机idation优化机制是怎样的,内容非常详细,感兴趣的小伙伴们可以参考借鉴,希望对大家能有所帮助。一、什么是Shuffle?
Shuffle是连接MapTask和ReduceTask过程的桥梁,MapTask的输出必须要经过Shuffle过程才能变成ReduceTask的输入,在分布式集群中,ReduceTask需要跨节点去拉取MapTask的输出结果,这中间涉及到数据的网络传输和磁盘IO,所以Shuffle的好坏将直接影响整个应用程序的性能,通常我们将Shuffle过程分成两部分:MapTask端的结果输出成为ShuffleWrite,ReduceTask端的数据拉取称为ShuffleRead。 Spark 的 Shuffle 过程与MapReduce的 Shuffle 过程原理基本相似,一些概念可直接套用,例如,Shuffle 过程中,提供数据的一端,被称作 Map 端,Map 端每个生成数据的任务称为 Mapper,对应的,接收数据的一端,被称作 Reduce 端,Reduce 端每个拉取数据的任务称为 Reducer,Shuffle 过程本质上都是将 Map 端获得的数据使用分区器进行划分,并将数据发送给对应的 Reducer 的过程。前面的文章我们已经讲过了Spark任务中Stage的划分依据是RDD的宽窄依赖;在窄依赖中父RDD和子RDD partition之间的关系是一对一的。或者父RDD一个partition只对应一个子RDD的partition情况下的父RDD和子RDD partition关系是多对一的。不会有shuffle的产生。父RDD的一个分区去到子RDD的一个分区。而在宽依赖中,父RDD与子RDD partition之间的关系是一对多。会有shuffle的产生。父RDD的一个分区的数据去到子RDD的不同分区里面。在现实场景中,百分之九十的调优情况都是发生在shuffle阶段,所以此类调优非常重要。二、普通的Spark HashShuffle原理
这里先看一张图,普通的Shuffle原理是怎么样的运行的,下面我会结合这个图给大家讲解基本的Shuffle原理:

普通Shuffle执行过程:
1.上面有三个ShuffleMapTask和两个ResultTask,ShuffleMapTask会根据ResultTask的数量创建出相应的bucket,bucket的数量是33。 2. 其次ShuffleMapTask产生的结果会根据设置的partition算法填充到每个bucket中去。这里的partition算法是可以自定义的,当然默认的算法是根据key哈希到不同的bucket中去,最后落地的话就是ShuffleBlockFIle。
3.ShuffleMapTask的输出的ShuffleBlockFIle位置信息作作为MapStatus发送到DAGScheduler的MapOutputTracker的Master中。
3.当ShuffleMapTask启动时,它会根据自己task的id和所依赖的ShuffleMapTask的id,然后去MapOutputTracker中读取ShuffleBlockFIle位置信息,最后从远端或是本地的blockmanager中取得相应的ShuffleBlockFIle作为ResultTask的输入进行处理。
如果ShuffleMapTask和ResultTask数量过多就会产生N*N的小文件,导致ShuffleWrite要耗费大量性能在磁盘文件的创建以及磁盘的IO上,给系统造成很大压力,上面我画的图有点不太好,这里我文字表述下: 一种情形A:如果有4个ShuffleMapTask和4个ResultTask,我的机器只有2个cpu核数,而每个task默认是需要一个cpu来运行的,这样我的4个ShuffleMapTask就分了两个批次运行,同时只有两个Task运行,第一批Task会生成2*4个ShuffleBlockFIle文件,第二批Task运行仍然会生成2*4的ShuffleBlockFIle文件,这样会产生16个小文件。 另一种情形B:我还是有4个ShuffleMapTask和4个ResultTask,我的机器只有4个cpu或者更多的cpu核数,我的4个ShuffleMapTask就会在同一个批次运行,还是会产生4*4=16个小文件。存在的问题:1.Shuffle前在磁盘上会产生海量的小文件,分布式模式ResultTask去拉取数据时,会产生大量会有过多的小文件创建和磁盘IO操作。2.可能导致OOM,大量耗时低效的 IO 操作 ,导致写磁盘时的对象过多,读磁盘时候的对象也过多,这些对象存储在堆内存中,会导致堆内存不足,相应会导致频繁的GC,GC会导致OOM。由于内存中需要保存海量文件操作句柄和临时信息,如果数据处理的规模比较庞大的话,内存不可承受,会出现 OOM 等问题。二、开启Consolidation机制的Spark HashShuffle原理鉴于上面基本Shuffle存在的不足,在后面的Spark0.81版本开始就引入了Consolidation机制,由参数spark.shuffle.consolidateFiles控制。将其设置为true即可开启优化机制,下面我们就看下优化后的Shuffle是如何处理的:
优化的Shuffle原理: 相当于对于上面的“情形B”做了优化,把在同一core上运行的多个ShuffleMapTask输出的合并到同一个文件,这样文件数目就变成了 cores*ResultTask个ShuffleBlockFile文件了,这里一定要注意同一个批次运行的ShuffleMapTask一定是写的不同的文件,只有不同批次的ShuffleMapTask才会写相同的文件,当第一批ShuffleMapTask运行完成后,后面在同一个cpu core上运行的TShuffleMapTask才会去写上一个在这个cpu core上运行ShuffleMapTask写的那个ShuffleBlockFile文件。至此Spark HashShuffle原理及其Consolidation机制讲解完毕,但是如果 Reducer 端的并行任务或者是数据分片过多的话则 Core * Reducer Task 依旧过大,也会产生很多小文件。上面的原理都是基于的HashShuffleManager。而Spark1.2.x以后,HashShuffleManager不再是Spark默认的Shuffle Manager,Spark1.2.x以后,Spark默认的Shuffle Manager是SortShuffleManager。在Spark2.0以后 HashShuffleManager已经被弃用。关于Shuffle原理及对应的Consolidation优化机制是怎样的就分享到这里了,希望以上内容可以对大家有一定的帮助,可以学到更多知识。如果觉得文章不错,可以把它分享出去让更多的人看到。

相关推荐: Harris角点的检测原理与流程介绍

本篇内容介绍了“Harris角点的检测原理与流程介绍”的有关知识,在实际案例的操作过程中,不少人都会遇到这样的困境,接下来就让小编带领大家学习一下如何处理这些情况吧!希望大家仔细阅读,能够学有所成!1. 角点概述角点是图像很重要的特征,对图像图形的理解和分析有…

免责声明:本站发布的图片视频文字,以转载和分享为主,文章观点不代表本站立场,本站不承担相关法律责任;如果涉及侵权请联系邮箱:360163164@qq.com举报,并提供相关证据,经查实将立刻删除涉嫌侵权内容。

(0)
打赏 微信扫一扫 微信扫一扫
上一篇 09/02 18:24
下一篇 09/02 18:24

相关推荐