SparkCore

键值对RDD数据分区器

Spark目前支持Hash分区和Range分区，用户也可以自定义分区，Hash分区为当前的默认分区，Spark中分区器直接决定了RDD中分区的个数、RDD中每条数据经过Shuffle过程属于哪个分区和Reduce的个数

注意：

(1)只有Key-Value类型的RDD才有分区器的，非Key-Value类型的RDD分区器的值是None

(2)每个RDD的分区ID范围：0~numPartitions-1，决定这个值是属于那个分区的。

获取RDD分区

可以通过使用RDD的partitioner 属性来获取 RDD 的分区方式。它会返回一个 scala.Option 对象，通过get方法获取其中的值。相关源码如下：

def getPartition(key: Any): Int = key match {
  case null => 0
  case _ => Utils.nonNegativeMod(key.hashCode, numPartitions)
}

def nonNegativeMod(x: Int, mod: Int): Int = {
  val rawMod = x % mod
  rawMod + (if (rawMod < 0) mod else 0)
}

（1）创建一个pairRDD

scala> val pairs = sc.parallelize(List((1,1),(2,2),(3,3)))
——>
pairs: org.apache.spark.rdd.RDD[(Int, Int)] = ParallelCollectionRDD[3] at parallelize at <console>:24

（2）查看RDD的分区器

scala> pairs.partitioner
——>
res1: Option[org.apache.spark.Partitioner] = None

（3）导入HashPartitioner类

scala> import org.apache.spark.HashPartitioner
import org.apache.spark.HashPartitioner

（4）使用HashPartitioner对RDD进行重新分区

scala> val partitioned = pairs.partitionBy(new HashPartitioner(2))
——>
partitioned: org.apache.spark.rdd.RDD[(Int, Int)] = ShuffledRDD[4] at partitionBy at <console>:27

（5）查看重新分区后RDD的分区器

scala> partitioned.partitioner
——>
res2: Option[org.apache.spark.Partitioner] = Some(org.apache.spark.HashPartitioner@2)

Hash分区

HashPartitioner分区的原理：对于给定的key，计算其hashCode，并除以分区的个数取余，如果余数小于0，则用余数+分区的个数（否则加0），最后返回的值就是这个key所属的分区ID。

使用Hash分区的实操

scala> nopar.partitioner
——>
res1: Option[org.apache.spark.Partitioner] = None

scala> val nopar = sc.parallelize(List((1,3),(1,2),(2,4),(2,3),(3,6),(3,8)),8)
——>
nopar: org.apache.spark.rdd.RDD[(Int, Int)] = ParallelCollectionRDD[10] at parallelize at <console>:24

scala>nopar.mapPartitionsWithIndex((index,iter)=>{ Iterator(index.toString+" : "+iter.mkString("|")) }).collect
——>
res2: Array[String] = Array("0 : ", 1 : (1,3), 2 : (1,2), 3 : (2,4), "4 : ", 5 : (2,3), 6 : (3,6), 7 : (3,8)) 

scala> val hashpar = nopar.partitionBy(new org.apache.spark.HashPartitioner(7))
——>
hashpar: org.apache.spark.rdd.RDD[(Int, Int)] = ShuffledRDD[12] at partitionBy at <console>:26

scala> hashpar.count
——>
res3: Long = 6

scala> hashpar.partitioner
——>
res4: Option[org.apache.spark.Partitioner] = Some(org.apache.spark.HashPartitioner@7)

scala> hashpar.mapPartitions(iter => Iterator(iter.length)).collect()
res19: Array[Int] = Array(0, 3, 1, 2, 0, 0, 0)