Spark RDD分区策略

RDD的数据分区策略由Partitioner数据分区器控制,Spark提供两个类型分片函数,如下:

《Spark RDD分区策略》 Partitioner类的代码依赖结构

Partitioner

《Spark RDD分区策略》 Partitioner

numPartitions:返回分区数量

key:根据key返回该key对应的分区编号,范围:[0, numPartitions-1]

HashPartitioner

基于哈希实现,对于给定的key,计算其hashCode,并除于分区的个数取余,如果余数小于0,则用余数+分区的个数,同时支持key值为null的情况,当key为null的时候,返回0,最后返回的值就是这个key所属的分区ID。

《Spark RDD分区策略》 HashPartitioner

若为负数则mod+numPartitions转为正数

《Spark RDD分区策略》 Utils#nonNegativeMod

RangePartitioner

HashPartitioner分区的实现原可能导致每个分区中数据量的不均匀,极端情况下会导致某些分区拥有RDD的全部数据(Hash冲突的原因)。

RangePartitioner分区则尽量保证每个分区中数据量的均匀,简单的说就是将一定范围内的数映射到某一个分区内。主要用于RDD的数据排序相关API中,比如sortByKey底层使用的数据分区器就是RangePartitioner分区器;该分区器的实现方式主要是通过两个步骤来实现的,第一步:先重整个RDD中抽取出样本数据,将样本数据排序,计算出每个分区的最大key值,形成一个Array[KEY]类型的数组变量rangeBounds;第二步:判断key在rangeBounds中所处的范围,给出该key值在下一个RDD中的分区id下标;该分区器要求RDD中的KEY类型必须是可以排序的,代码说明如下:

《Spark RDD分区策略》 排序器
《Spark RDD分区策略》 RangePartitioner#rangeBounds

119~120:分区数是一个的情况下,直接返回一个空的集合,表示数据不进行分区

123:数据抽样大小,最多1M的数据量(10^6),最少20倍的RDD分区数量,也就是每个RDD分区至少抽取20条数据

125:计算每个分区抽取的数据量大小,假设输入数据每个分区分布的比较均匀。对于超大数据集(分区数超过5万的)乘以3会让数据稍微增大一点,对于分区数低于5万的数据集,每个分区抽取数据量为60条。5万是10^6 / 20得出,60是20 * 3的得出。

126:从rdd中抽取数据,返回值:(总rdd数据量, Array[分区id,当前分区的数据量,当前分区抽取的数据])。sketch函数对父RDD中的每个分区进行采样,并记录下分区的ID和分区中数据总和。

《Spark RDD分区策略》 RangePartitioner#sketch

128:如果总的数据量为0(RDD为空),那么直接返回一个空的数组

132:计算总样本数量和总记录数的占比,占比最大为1.0

133:保存样本数据的集合buffer

134:保存数据分布不均衡的分区id(数据量超过fraction比率的分区)

135:计算抽取出来的样本数据

136:如果fraction乘以当前分区中的数据量大于之前计算的每个分区的抽象数据大小,那么表示当前分区抽取的数据太少了,该分区数据分布不均衡,需要重新抽取

140:当前分区不属于数据分布不均衡的分区,计算占比权重,并添加到candidates集合中

146:对于数据分布不均衡的RDD分区,重新进行数据抽样

148:获取数据分布不均衡的RDD分区,并构成RDD

149:随机种子

150:利用rdd的sample抽样函数API进行数据抽样

154:将最终的抽样数据计算出rangeBounds出来

《Spark RDD分区策略》 rangeBounds#getPartition

159:下一个RDD的分区数量是rangeBounds数组中元素数量+ 1个

161:二分查找器

163:根据RDD的key值返回对应的分区id,从0开始

164:强制转换key类型为RDD中原本的数据类型

168:如果分区数据小于等于128个,那么直接本地循环寻找当前k所属的分区下标

173:如果分区数量大于128个,那么使用二分查找方法寻找对应k所属的下标

175:但是如果k在rangeBounds中没有出现,实质上返回的是一个负数(范围)或者是一个超过rangeBounds大小的数(最后一个分区,比所有数据都大)

182:根据数据排序是升序还是降序进行数据的排列,默认为升序

RangePartitioner构建rangeBounds数组对象,主要步骤是:

1. 如果分区数量小于2或者rdd中不存在数据的情况下,直接返回一个空的数组,不需要计算range的边界;如果分区数据大于1的情况下,而且rdd中有数据的情况下,才需要计算数组对象

2. 计算总体的数据抽样大小sampleSize,计算规则是:至少每个分区抽取20个数据或者最多1M的数据量

3. 根据sampleSize和分区数量计算每个分区的数据抽样样本数量sampleSizePrePartition

4. 调用RangePartitioner的sketch函数进行数据抽样,计算出每个分区的样本

5. 计算样本的整体占比以及数据量过多的数据分区,防止数据倾斜

6. 对于数据量比较多的RDD分区调用RDD的sample函数API重新进行数据抽取

7. 将最终的样本数据通过RangePartitoner的determineBounds函数进行数据排序分配,计算出rangeBounds

getPartition定位分区ID,算法相对简单:如果分区边界数组的大小小于或等于128的时候直接变量数组,否则采用二分查找法确定key属于某个分区。

从上面的采样算法可以看出,对于不同的分区weight的值是不一样的,这个值对应的就是每个分区的采样间隔。

《Spark RDD分区策略》 RangePartitioner#determineBounds

这个函数最后返回的就是分区的划分边界。

总结

基本上HashPartitioner已经满足绝大部分需求,RangePartitioner从上述分析使用场景有一定的局限。

另外,在特殊情况下用户可自定义Partitioner,只需要扩展Partitioner抽象类,实现下面3个方法:

def numPartitions: Int:这个方法需要返回你想要创建分区的个数;

def getPartition(key: Any): Int:这个函数需要对输入的key做计算,然后返回该key的分区ID,范围一定是0到numPartitions-1;

equals():这个是Java标准的判断相等的函数,之所以要求用户实现这个函数是因为Spark内部会比较两个RDD的分区是否一样。

    原文作者:木戎
    原文地址: https://www.jianshu.com/p/a21b3be88afd
    本文转自网络文章,转载此文章仅为分享知识,如有侵权,请联系博主进行删除。
点赞