Android WorkManager使用入门

注册谷歌账号遇到手机号无法验证的解决方法

  返回  

大数据之Spark(六):Spark Shuffle 原理

2021/8/21 21:41:52 浏览:

Spark Shuffle分为Write和Read两个过程。

在Spark中负责shuffle过程的执行、计算、处理的组件主要是 ShuffleManager,其是一个trait,负责管理本地以及远程的block数据的shuffle操作。

所有方法如下图所示:

ShuffleManager定义的方法

 由SparkEnv的shuffleManager管理

主要方法解释:

  • registerShuffle:注册ShuffleDependency,同时获取一个ShuffleHandle(后面根据ShuffleHandle来决定使用不同的ShuffleWrite)并将其传递给任务。
    ps:注册时机:由ShuffledRDD的getDependencies方法创建ShuffleDependency,创建时即注册该ShuffleDenpendency。

  • getWriter:返回ShuffleWriter用于Shuffle Write过程。对一个分区返回一个ShuffleWriter,并由executors上的ShuffleMapTask任务调用(Spark中的任务分为两种:ShuffleMapTask与ResultTask)。

  • getReader:返回ShuffleReader用于Shuffle Read过程。

在早期版本中,ShuffleManager的实现者是HashShuffleManager,而新版本中只有SortShuffleManager。

前者存在的问题:会产生大量的磁盘文件,进而有大量的磁盘 IO 操作,比较影响性能。SortShuffleManager 相对来说,有了一定的改进。主要就在于,每个 Task 在 Shuffle Write 操作时,虽然也会产生较大的磁盘文件,但最后会将所有的临时文件合并 (merge) 成一个磁盘文件,因此每个 Task 就只有一个磁盘文件。在下一个 Stage 的 Shuffle Read Task 拉取自己数据的时候,只要根据索引拉取每个磁盘文件中的部分数据即可。

一、Shuffle Write

何时发生Shuffle Write?
Shuffle Write操作发生在ShuffleMapTask#runTask中,代码如下图所示:

ShuffleMapTask #runTask

Shuffle Write的处理逻辑会放到该ShuffleMapStage的最后,因为rdd.iterator()将调用compute方法,并且递归调用父RDD的compute方法。如下两图所示:

RDD #iterator
RDD #computeOrReadCheckpoint
MapPartitionsRDD #compute

 

因为Spark以Shuffle发生与否来划分Stage,所以该Stage的final RDD每输出一个 record 就将其分区并持久化。
ShuffleWriter有三种实现:

  • BypassMergeSortShuffleWriter:Hash风格的基于Sort的Shuffle机制,和已经废弃的HashShuffleWriter类似
  • UnsafeShuffleWriter:支持relocation,可以对已经序列化的对象进行排序,这种排序起到的效果和先对数据排序再序列化一致。支持relocation的Serializer是KryoSerializer,Spark默认使用JavaSerializer,通过参数spark.serializer设置(SparkConf.set("spark.serializer", "org.apache.spark.serializer.KryoSerializer"))
  • SortShuffleWriter

 如图所示:

ShuffleWriter的多种实现

 

上图中三种Writer又分别对应不同的Handle:

  • BypassMergeSortShuffleWriter-> BypassMergeSortShuffleHandle
  • UnsafeShuffleWriter-> SerializedShuffleHandle
  • SortShuffleWriter-> BaseShuffleHandle

 使用哪种ShuffleWrite?

根据ShuffleHandle来决定使用不同的ShuffleWrite,如下图所示:

org.apache.spark.shuffle.sort.SortShuffleManager #getWriter

何时指定ShuffleHandle?

在构建ShuffleDependency时会构建ShuffleHandle。创建时在registerShuffle方法中,有着对ShuffleHandle使用的条件约束,如下图所示:

 

SortShuffleManager#registerShuffle

 1)如果分区小于spark.shuffle.sort.bypassMergeThreshold(默认200),且map端没有聚合操作,使用BypassMergeSortShuffleHandle,否则进入下一个条件。如下图所示:
 

org.apache.spark.shuffle.sort.SortShuffleWriter #shouldBypassMergeSort

 2)如果map端没有聚合操作,且Serializer支持重定位(即使用KryoSerializer),且分区数目小于16777216(最大分区号)时使用SerializedShuffleHandle。否则进入下一条件。如下图所示:
 

SortShuffleManager #canUseSerializedShuffle

 3)以上条件都不满足时使用BaseShuffleHandle。对应的ShuffleWrite是SortShuffleWriter,这种形式的支持map端聚合操作,同时支持排序。这种是最通用的Writer。早期的HashShuffleWriter的主要弊端是产生的临时文件太多,那么Sort ShuffleWriter使相同的ShuffleMapTask 公用一个输出文件,然后创建一个索引文件对这个文件进行索引,如下图所示:

 

早期基于Hash的Shuffle写操作
基于排序的Shuffle写操作

 二、Shuffle Read

何时发生Shuffle Read?
Shuffle Read操作发生在ShuffledRDD#compute方法中,意味着Shuffle Read可以发生在ShuffleMapTask(非最后一个阶段的任务)和ResultTask(最后一个阶段的任务)两种任务中。代码如下图所示:
 

ShuffledRDD #compute

 

每个Stage的上边界,可以从外部存储读取数据,也或者是需要读取上一个Stage的输出,而下边界要么需要写入本地文件系统(Shuffle),以供下一个Stage读取,要么是最后一个Stage,需要输出结果。
除了需要从外部存储读取数据和RDD已经持久化(Cache、Checkpoint),一般Task都是从ShuffledRDD的Shuffle Read开始的。
ShuffleManager#getReader实例化一个BlockStoreShuffleReader。如下图所示:

 

ShuffleManager #getReader

 BlockStoreShuffleReader#read首先实例化了ShuffleBlockFetcherIterator对象,如下图所示:

BlockStoreShuffleReader #read

上图中“mapOutputTracker.getMapSizesByExecutorId”返回存储数据位置的元数据。类的定义如下图所示:

ShuffleBlockFetcherIterator类定义

blocksByAddress指出了数据是来自于哪个节点的哪些block,并且block的数据大小是多少。接下来初始化代码如下所示:

 

ShuffleBlockFetcherIterator初始化

从代码可以看出:
1)首先区分是本地还是远程blocks,返回远程请求FetchRequest加入到fetchRequests队列中。2)从fetchRequests取出远程请求,并使用sendRequest方法发送请求,获取远程数据。
3)获取本地blocks。

  •  本地读取

fetchLocalBlocks()负责本地blocks的获取,在上图的splitLocalRemoteBlocks中,已经将本地的blocks列表存入了localBlocks。如下图所示:
 

org.apache.spark.storage.ShuffleBlockFetcherIterator#fetchLocalBlocks
  •  远程读取

 fetchUpToMaxBytes()负责远程数据读取

org.apache.spark.storage.ShuffleBlockFetcherIterator #fetchUpToMaxBytes

 从fetchRequests中取出FetchRequest,并调用了sendRequest()方法。sendRequest()向远程节点发起读取block的请求。关键代码如下图所示:

org.apache.spark.storage.ShuffleBlockFetcherIterator #sendRequest

 blockFetchingListener定义了一个监听器读取数据

sendRequest()关键代码 - 1

 shuffleClient.fetchBlocks()封装了对远程数据块的读取,由blockFetchingListener接收结果数据:见上图result.put()代码

sendRequest()关键代码 - 2

 

 

 

 

 

联系我们

如果您对我们的服务有兴趣,请及时和我们联系!

服务热线:18288888888
座机:18288888888
传真:
邮箱:888888@qq.com
地址:郑州市文化路红专路93号