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

Partitioner类的代码依赖结构

Partitioner

Partitioner

numPartitions：返回分区数量

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

HashPartitioner

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

HashPartitioner

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

Utils#nonNegativeMod

RangePartitioner

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

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

排序器 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和分区中数据总和。

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出来

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的值是不一样的，这个值对应的就是每个分区的采样间隔。

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的分区是否一样。