Spark RDD分区策略
RDD的数据分区策略由Partitioner数据分区器控制,Spark提供两个类型分片函数,如下:
Partitioner类的代码依赖结构Partitioner
PartitionernumPartitions:返回分区数量
key:根据key返回该key对应的分区编号,范围:[0, numPartitions-1]
HashPartitioner
基于哈希实现,对于给定的key,计算其hashCode,并除于分区的个数取余,如果余数小于0,则用余数+分区的个数,同时支持key值为null的情况,当key为null的时候,返回0,最后返回的值就是这个key所属的分区ID。
HashPartitioner若为负数则mod+numPartitions转为正数
Utils#nonNegativeModRangePartitioner
HashPartitioner分区的实现原可能导致每个分区中数据量的不均匀,极端情况下会导致某些分区拥有RDD的全部数据(Hash冲突的原因)。
RangePartitioner分区则尽量保证每个分区中数据量的均匀,简单的说就是将一定范围内的数映射到某一个分区内。主要用于RDD的数据排序相关API中,比如sortByKey底层使用的数据分区器就是RangePartitioner分区器;该分区器的实现方式主要是通过两个步骤来实现的,第一步:先重整个RDD中抽取出样本数据,将样本数据排序,计算出每个分区的最大key值,形成一个Array[KEY]类型的数组变量rangeBounds;第二步:判断key在rangeBounds中所处的范围,给出该key值在下一个RDD中的分区id下标;该分区器要求RDD中的KEY类型必须是可以排序的,代码说明如下:
排序器 RangePartitioner#rangeBounds119~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和分区中数据总和。
RangePartitioner#sketch128:如果总的数据量为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出来
rangeBounds#getPartition159:下一个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的值是不一样的,这个值对应的就是每个分区的采样间隔。
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的分区是否一样。