Spark RDD-Operations

键值对泛型的 RDD 可以通过 partitionBy 指定分区器。

Spark 默认实现了两种分区器：HashPartitioner、RangePartitioner，也可以自定义分区器。

• HashPartitioner：根据 key 的 hashCode 返回值对分区数取模
  • 优势：可以将相同 key 的元素分到同一分区，方便 byKey 的操作
  • 劣势：如果某些相同 key 的元素较多，容易造成数据倾斜
• RangePartitioner：使用抽样方法，随机抽样并轮询分发数据到不容的分区
  • 优势：发分区后每个RDD中的元素数量相差无几
  • 劣势：会将数据打乱，如需 byBey 操作会重新进行 shuffle

自定义分区器：

• 使用匿名类的方式自定义分区器

rdd.map(x => (x,x))
.partitionBy(new Partitioner {
    // 设置分区数量
    override def numPartitions: Int = 2
    // 根据 key 计算出分区编号
    override def getPartition(key: Any): Int = key.asInstanceOf[Int] % 2
})

• 单独构造类的方式自定义分区器
  • 继承 Partitioner 抽象类
  • 重写其中的 numPartitions、getPartition 分发

rdd.map(x => (x, x))
.partitionBy(new DefinePartition(2))

class DefinePartition(num: Int) extends Partitioner {
    override def numPartitions: Int = num
    override def getPartition(key: Any): Int = key.asInstanceOf[Int] % 2
}

使用 mapPartitionsWithIndex 方法，还能获取分区编号

在上面可以看到很多 ByKey 的算子，这些算子都适配于 (K, V) 泛型的 RDD，此类算子底层都是基于 combineByKeyWithClassTag 实现。

combineByKeyWithClassTag 中的 WithClassTag 相当于一种泛型检测机制，在该算子之上有一个简单的继承，名为combineByKey。

def combineByKey[C](
    createCombiner: V => C,
    mergevalue: (C, V) => C,
    mergeCombiners: (C, C) => C,
    partitioner: Partitioner,
    mapSideCombine: Boolean = true,
    serializer: Serializer = null): RDD[(K, C)] = self.withScope {
    combineByKeyWithClassTag(createCombiner, mergevalue, mergeCombiners,
                             partitioner, mapSideCombine, serializer)(null)
}

combineByKey 还是在调用 combineByKeyWithClassTag，为了方便，看此方法就可以了！

 combineByKey 有3个重要的参数：

• createCombiner: V => C：拿到第一个value，创建一个（任意类型的）聚合器对象
• mergevalue: (C, V) => C：将 value 与上一次计算完的 C 进行合并，如果拿到 value 没有聚合器，则先进行 createCombiner
• mergeCombiners: (C, C) => C：将多个 C 进行合并

 例子：

rdd = sc.makeRDD(Seq(1, 1, 1, 2, 2, 3))
rdd.map(x => (x, 1))  // 此时 value 的类型为 Int
.combineByKey[Double](
    (V: Int) => V.toDouble,
    (C: Double, V: Int) => C + V,
    (C1: Double, C2: Double) => C1 + C2)
.foreach(println)

// 输出： 可以看到 value 的类型变为了 Double
(1,3.0)
(2,2.0)
(3,1.0)

 combineByKey 与 reduceByKey 的区别是前者聚合完可以改变 value 的类型，而后者不行；reduceByKey 实则就是调用 combineByKey，只不过不改变 V 的类型！如下源码：

def reduceByKey(
    partitioner: Partitioner, 
    func: (V, V) => V): RDD[(K, V)] = self.withScope {
    combineByKeyWithClassTag[V]((v: V) => v, func, func, partitioner) // value 的类型并没有改变
}