Spark快速上手(4)Spark核心编程-Spark分区器(Partitioner)@(RDD-K_V) ...

@Spark分区器(Partitioner)

HashPartitioner（默认的分区器）

HashPartitioner分区原理是对于给定的key，计算其hashCode，并除以分区的个数取余，如果余数小于0，则余数+分区的个数，最后返回的值就是这个key所属的分区ID，当key为null值是返回0。
源码在org.apache.spark包下：
origin code:

1. class HashPartitioner(partitions: Int) extends Partitioner {
2.   require(partitions >= 0, s"Number of partitions ($partitions) cannot be negative.")
4.   def numPartitions: Int = partitions
5.   // 根据键的值来判断在哪一个分区
6.   def getPartition(key: Any): Int = key match {
7.     case null => 0   // 键为null始终在0分区
8.     case _ => Utils.nonNegativeMod(key.hashCode, numPartitions) // 键不为0，根据键的hashCode值和分区数进行计算
9.   }
10.   
11.   override def equals(other: Any): Boolean = other match {
12.     case h: HashPartitioner =>
13.       h.numPartitions == numPartitions
14.     case _ =>
15.       false
16.   }
18. …………
19. }
20. // 底层实质：取模运算
21. def nonNegativeMod(x: Int, mod: Int): Int = {
22.    val rawMod = x % mod
23.    rawMod + (if (rawMod < 0) mod else 0)
24. }

复制代码

RangePartitioner

HashPartitioner分区的实现可能会导致数据倾斜，极端情况下会导致某些分区拥有RDD的所有数据。而RangePartitioner分区器则尽量保证各个分区数据均匀，而且分区和分区之间是有序的，也就是说令一个分区中的元素均比另一个分区中的元素小或者大；但是分区内的元素是不能保证顺序的。简单地说就是将一定范围内的数据映射到一个分区内。
  sortByKey底层使用的数据分区器就是RangePartitioner分区器，该分区器的实现方式主要通过两个步骤实现：
①先从整个RDD中抽取样本数据，将样本数据排序，计算出每个分区的最大key值，形成一个Array[key]类型的数组变量rangeBounds；
②判断key在rangeBounds中所处的范围，给出该key值在下一个RDD中的分区id下标。该分区器要求RDD中的key类型必须是可排序的。
origin code:

1. class RangePartitioner[K : Ordering : ClassTag, V](
2.     partitions: Int,
3.     rdd: RDD[_ <: Product2[K, V]],
4.     private var ascending: Boolean = true,
5.     val samplePointsPerPartitionHint: Int = 20)
6.   extends Partitioner {
8.   // A constructor declared in order to maintain backward compatibility for Java, when we add the
9.   // 4th constructor parameter samplePointsPerPartitionHint. See SPARK-22160.
10.   // This is added to make sure from a bytecode point of view, there is still a 3-arg ctor.
11.   def this(partitions: Int, rdd: RDD[_ <: Product2[K, V]], ascending: Boolean) = {
12.     this(partitions, rdd, ascending, samplePointsPerPartitionHint = 20)
13.   }
15.   // We allow partitions = 0, which happens when sorting an empty RDD under the default settings.
16.   require(partitions >= 0, s"Number of partitions cannot be negative but found $partitions.")
17.   require(samplePointsPerPartitionHint > 0,
18.     s"Sample points per partition must be greater than 0 but found $samplePointsPerPartitionHint")
20.   // 获取RDD中key类型数据的排序器
21.   private var ordering = implicitly[Ordering[K]]
23.   // An array of upper bounds for the first (partitions - 1) partitions
24.   private var rangeBounds: Array[K] = {
25.     if (partitions <= 1) {
26.       // 如果给定的分区数是一个的情况下，直接返回一个空的集合，表示数据不进行分区
27.       Array.empty
28.     } else {
29.       // This is the sample size we need to have roughly balanced output partitions, capped at 1M.
30.       // Cast to double to avoid overflowing ints or longs
31.       // 给定总的数据抽样大小，最多1M的数据量（10^6），最少20倍的RDD分区数量，也就是每个RDD分区至少抽取20条数据
32.       val sampleSize = math.min(samplePointsPerPartitionHint.toDouble * partitions, 1e6)
33.       // Assume the input partitions are roughly balanced and over-sample a little bit.
34.       // 计算每个分区抽样的数据量大小，假设输入数据每个分区分布的比较均匀
35.       // 对于超大数据集（分区数量超过5万的）乘以3会让数据稍微增大一点，对于分区数低于5万的数据集，每个分区抽取数据量为60条也不算多
36.       val sampleSizePerPartition = math.ceil(3.0 * sampleSize / rdd.partitions.length).toInt
37.       // 从RDD中抽取数据，返回值：（总RDD数据量，Array[分区id, 当前分区的数据量, 当前分区抽取的数据]）
38.       val (numItems, sketched) = RangePartitioner.sketch(rdd.map(_._1), sampleSizePerPartition)
39.       if (numItems == 0L) {
40.         // 如果总的数据量为0（RDD为空），那么直接返回一个空的数组
41.         Array.empty
42.       } else {
43.         // If a partition contains much more than the average number of items, we re-sample from it
44.         // to ensure that enough items are collected from that partition.
45.         // 计算总样本数量和总记录数的占比，占比最大为1.0
46.         val fraction = math.min(sampleSize / math.max(numItems, 1L), 1.0)
47.         // 保存样本数据的集合buffer
48.         val candidates = ArrayBuffer.empty[(K, Float)]
49.         // 保存数据分布不均衡的分区id（数据量超过fraction比率的分区）
50.         val imbalancedPartitions = mutable.Set.empty[Int]
51.         // 计算抽取出来的样本数据
52.         sketched.foreach { case (idx, n, sample) =>
53.           if (fraction * n > sampleSizePerPartition) {
54.             // 如果fraction乘以当前分区中的数据量大于之前计算的每个分区的抽样数据大小，那么表示当前分区抽取的数据太少了，该分区数据分布不均衡，需要重新抽取
55.             imbalancedPartitions += idx
56.           } else {
57.             // 当前分区不属于数据分布不均衡的分区，计算占比权重，并添加到candidates集合中
58.             // The weight is 1 over the sampling probability.
59.             val weight = (n.toDouble / sample.length).toFloat
60.             for (key <- sample) {
61.               candidates += ((key, weight))
62.             }
63.           }
64.         }
65.         // 对数据分布不均衡的RDD分区，重新进行数据抽样
66.         if (imbalancedPartitions.nonEmpty) {
67.           // Re-sample imbalanced partitions with the desired sampling probability.
68.           // 获取数据分布不均衡的RDD分区，并构成RDD
69.           val imbalanced = new PartitionPruningRDD(rdd.map(_._1), imbalancedPartitions.contains)
70.           // 随机种子
71.           val seed = byteswap32(-rdd.id - 1)
72.           // 利用RDD的sample抽样函数API进行数据抽样
73.           val reSampled = imbalanced.sample(withReplacement = false, fraction, seed).collect()
74.           val weight = (1.0 / fraction).toFloat
75.           candidates ++= reSampled.map(x => (x, weight))
76.         }
77.         // 将最终的抽样数据计算出rangeBounds
78.         RangePartitioner.determineBounds(candidates, math.min(partitions, candidates.size))
79.       }
80.     }
81.   }
83.   // 下一个RDD的分区数量是rangeBounds数组中元素数量+1个
84.   def numPartitions: Int = rangeBounds.length + 1
86.   // 二分查找器，内部使用Java中的Arrays提供的二分查找方法
87.   private var binarySearch: ((Array[K], K) => Int) = CollectionsUtils.makeBinarySearch[K]
89.   // 根据RDD的key值返回对应的分区id，从0开始
90.   def getPartition(key: Any): Int = {
91.     // 强制转换key类型为RDD中原本的数据类型
92.     val k = key.asInstanceOf[K]
93.     var partition = 0
94.     if (rangeBounds.length <= 128) {
95.       // If we have less than 128 partitions naive search
96.       // 如果分区数据小于等于128个，那么直接本地循环寻找当前k所属的分区下标
97.       while (partition < rangeBounds.length && ordering.gt(k, rangeBounds(partition))) {
98.         partition += 1
99.       }
100.     } else {
101.       // Determine which binary search method to use only once.
102.       // 如果分区数量大于128个，那么使用二分查找方法寻找对应k所属的下标
103.       // 但是如果k在rangeBounds中没有出现，实质上返回的是一个负数（范围）或者是一个超过rangeBounds大小的数（最后一个分区，比所有的数据都大）
104.       partition = binarySearch(rangeBounds, k)
105.       // binarySearch either returns the match location or -[insertion point]-1
106.       if (partition < 0) {
107.         partition = -partition-1
108.       }
109.       if (partition > rangeBounds.length) {
110.         partition = rangeBounds.length
111.       }
112.     }
113.     // 根据数据排序是升序还是降序进行数据的排列，默认为升序
114.     if (ascending) {
115.       partition
116.     } else {
117.       rangeBounds.length - partition
118.     }
119.   }
121.   override def equals(other: Any): Boolean = other match {
122.     case r: RangePartitioner[_, _] =>
123.       r.rangeBounds.sameElements(rangeBounds) && r.ascending == ascending
124.     case _ =>
125.       false
126.   }
128.   override def hashCode(): Int = {
129.     val prime = 31
130.     var result = 1
131.     var i = 0
132.     while (i < rangeBounds.length) {
133.       result = prime * result + rangeBounds(i).hashCode
134.       i += 1
135.     }
136.     result = prime * result + ascending.hashCode
137.     result
138.   }
140.   @throws(classOf[IOException])
141.   private def writeObject(out: ObjectOutputStream): Unit = Utils.tryOrIOException {
142.     val sfactory = SparkEnv.get.serializer
143.     sfactory match {
144.       case js: JavaSerializer => out.defaultWriteObject()
145.       case _ =>
146.         out.writeBoolean(ascending)
147.         out.writeObject(ordering)
148.         out.writeObject(binarySearch)
150.         val ser = sfactory.newInstance()
151.         Utils.serializeViaNestedStream(out, ser) { stream =>
152.           stream.writeObject(scala.reflect.classTag[Array[K]])
153.           stream.writeObject(rangeBounds)
154.         }
155.     }
156.   }
158.   @throws(classOf[IOException])
159.   private def readObject(in: ObjectInputStream): Unit = Utils.tryOrIOException {
160.     val sfactory = SparkEnv.get.serializer
161.     sfactory match {
162.       case js: JavaSerializer => in.defaultReadObject()
163.       case _ =>
164.         ascending = in.readBoolean()
165.         ordering = in.readObject().asInstanceOf[Ordering[K]]
166.         binarySearch = in.readObject().asInstanceOf[(Array[K], K) => Int]
168.         val ser = sfactory.newInstance()
169.         Utils.deserializeViaNestedStream(in, ser) { ds =>
170.           implicit val classTag = ds.readObject[ClassTag[Array[K]]]()
171.           rangeBounds = ds.readObject[Array[K]]()
172.         }
173.     }
174.   }
175. }

复制代码

将一定范围内的数映射到某一个分区内，在实现中，分界（rangeBounds）算法用到了水塘抽样算法。RangePartitioner的重点在于构建rangeBounds数组对象，主要步骤是：

• 如果分区数量小于2或者RDD中不存在数据的情况下，直接返回一个空的数组，不需要计算range的边界；如果分区数量大于1的情况下，而且RDD中有数据的情况下，才需要计算数组对象
  
• 计算总体的数据抽样大小sampleSize，计算规则是：至少每个分区抽取20个数据或者最多1M的数据量
  
• 根据sampleSize和分区数量计算每个分区的数据抽样样本数量sampleSizePartition
  
• 调用RangePartitioner的sketch函数进行数据抽样，计算出每个分区的样本
  
• 计算样本的整体占比以及数据量过多的数据分区，防止数据倾斜
  
• 对于数据量比较多的RDD分区调用RDD的sample函数API重新进行数据获取
  
• 将最终的样本数据通过RangePartitioner的determineBounds函数进行数据排序分配，计算出rangeBounds
  

RangePartitioner的sketch函数的作用是对RDD中的数据按照需要的样本数据量进行数据抽取，主要调用SamplingUtils类的reservoirSampleAndCount方法对每个分区进行数据抽取，抽取后计算出整体所有分区的数据量大小；reserviorSampleAndCount方法的抽取方式是先从迭代器中获取样本数量个数据（顺序获取），然后对剩余的数据进行判断，替换之前的样本数据，最终达到数据抽样的效果。RangePartitioner的determineBounds函数的作用是根据样本数据记忆权重大小确定数据边界。
RangePartitioner的determineBounds函数的作用是根据样本数据记忆权重大小确定数据边界，源代码如下：
origin code:

1. /**
2.    * Determines the bounds for range partitioning from candidates with weights indicating how many
3.    * items each represents. Usually this is 1 over the probability used to sample this candidate.
4.    *
5.    * @param candidates unordered candidates with weights
6.    * @param partitions number of partitions
7.    * @return selected bounds
8.    */
9.   def determineBounds[K : Ordering : ClassTag](
10.       candidates: ArrayBuffer[(K, Float)],
11.       partitions: Int): Array[K] = {
12.     val ordering = implicitly[Ordering[K]]
13.     // 按照数据进行排序，默认升序排序
14.     val ordered = candidates.sortBy(_._1)
15.     // 获取总的样本数据大小
16.     val numCandidates = ordered.size
17.     // 计算总的权重大小
18.     val sumWeights = ordered.map(_._2.toDouble).sum
19.     // 计算步长
20.     val step = sumWeights / partitions
21.     var cumWeight = 0.0
22.     var target = step
23.     val bounds = ArrayBuffer.empty[K]
24.     var i = 0
25.     var j = 0
26.     var previousBound = Option.empty[K]
27.     while ((i < numCandidates) && (j < partitions - 1)) {
28.       // 获取排序后的第i个数据及权重
29.       val (key, weight) = ordered(i)
30.       // 累计权重
31.       cumWeight += weight
32.       if (cumWeight >= target) {
33.         // Skip duplicate values.
34.         // 权重已经达到一个步长的范围，计算出一个分区id的值
35.         if (previousBound.isEmpty || ordering.gt(key, previousBound.get)) {// 上一个边界值为空，或者当前边界值key数据大于上一个边界的值，那么当前key有效，进行计算
36.           // 添加当前key到边界集合中
37.           bounds += key
38.           // 累计target步长界限
39.           target += step
40.           // 分区数量加1
41.           j += 1
42.           // 上一个边界的值重置为当前边界的值
43.           previousBound = Some(key)
44.         }
45.       }
46.       i += 1
47.     }
48.     // 返回结果
49.     bounds.toArray
50.   }

复制代码

自定义分区器

自定义分区器是需要继承org.apache.spark.Partitioner类并实现以下三个方法：

• numPartitioner: Int：返回创建出来的分区数
  
• getPartition(key: Any): Int：返回给定键的分区编号（0到numPartitions - 1）
  
• equals()：Java判断相等性的标准方法。这个方法的实现非常重要，Spark需要用这个方法来检查你的分区器是否和其他分区器实例相同，这样Spark才可以判断两个RDD的分区方式是否相同
  

e.g.1

1. // CustomPartitioner
2. import org.apache.spark.Partitioner
4. /**
5. * @param numPartition 分区数量
6. */
7. class CustomPartitioner(numPartition: Int) extends Partitioner{
8.     // 返回分区的总数
9.     override def numPartitions: Int = numPartition
11.     // 根据传入的 key 返回分区的索引
12.     override def getPartition(key: Any): Int = {
13.         key.toString.toInt % numPartition
14.     }
15. }
17. // CustomPartitionerDemo
18. import com.work.util.SparkUtil
19. import org.apache.spark.SparkContext
20. import org.apache.spark.rdd.RDD
23. object CustomPartitionerDemo {
24.     def main(args: Array[String]): Unit = {
25.         val sc: SparkContext = SparkUtil.getSparkContext()
26.         println("=================== 原始数据 =====================")
27.         // zipWithIndex 该函数将 RDD 中的元素和这个元素在 RDD 中的 ID（索引号）组合成键值对
28.         val data: RDD[(Int, Long)] = sc.parallelize(0 to 10, 1).zipWithIndex()
29.         println(data.collect().toBuffer)
31.         println("=================== 分区和数据组合成 Map =====================")
32.         val func: (Int, Iterator[(Int, Long)]) => Iterator[String] = (index: Int, iter: Iterator[(Int, Long)]) => {
33.             iter.map(x => "[partID:" + index + ", value:" + x + "]")
34.         }
35.         val array: Array[String] = data.mapPartitionsWithIndex(func).collect()
36.         for (i <- array) {
37.             println(i)
38.         }
40.         println("=================== 自定义5个分区和数据组合成 Map =====================")
41.         val rdd1: RDD[(Int, Long)] = data.partitionBy(new CustomPartitioner(5))
42.         val array1: Array[String] = rdd1.mapPartitionsWithIndex(func).collect()
43.         for (i <- array1) {
44.             println(i)
45.         }
46.     }
47. }

复制代码

免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！