spark API 之 combineByKey

import org.apache.spark.{HashPartitioner, SparkConf, SparkContext}

/**
* Created by luckuan on 16/4/19.
*/
object TT {
def main(args: Array[String]) {

val sparkConf = new SparkConf()
sparkConf.setMaster("local[*]")
//    sparkConf.set("spark.serializer", "org.apache.spark.serializer.KryoSerializer")
sparkConf.setAppName("ts")
val sc: SparkContext = new SparkContext(sparkConf)

val partitionNum = 2

val rdd = sc.makeRDD(Seq(("a1", "11"), ("a1", "1"), ("a1", "111"), ("a2", "2"), ("a2", "22"), ("a3", "3"), ("a4", "4"), ("a1", "1111"), ("a2", "222")), partitionNum)

val zero = () => scala.collection.mutable.HashSet.empty[String]
val seq = (s: scala.collection.mutable.HashSet[String], v: String) => {
val s_clone = s.clone() //此处clone是为了记录原有的值,否则在下次打印的时候是最终结果,不太直观,线上不需要用到clone.
val ret = s += v
println(s"seq-s:${s_clone}--seq-v:${v}---seq-rs:${ret}")
ret
}
val comb = (u: scala.collection.mutable.HashSet[String], v: scala.collection.mutable.HashSet[String]) => {
{
val u_clone = u.clone()
val ret = u ++ v
println(s"comb-s:${u_clone}--comb-v:${v}---comb-rs:${ret}")
ret
}
}

val ret = rdd.combineByKey((v: String) => {
println(s"根据[${v}]进行初始化")
seq(zero(), v)
}, seq, comb, new HashPartitioner(partitionNum)).collect()

ret.foreach(println)

}
}

RDD分区为1

根据[11]进行初始化
seq-s:Set()--seq-v:11---seq-rs:Set(11)
seq-s:Set(11)--seq-v:1---seq-rs:Set(1, 11)
seq-s:Set(1, 11)--seq-v:111---seq-rs:Set(1, 111, 11)
根据[2]进行初始化
seq-s:Set()--seq-v:2---seq-rs:Set(2)
seq-s:Set(2)--seq-v:22---seq-rs:Set(2, 22)
根据[3]进行初始化
seq-s:Set()--seq-v:3---seq-rs:Set(3)
根据[4]进行初始化
seq-s:Set()--seq-v:4---seq-rs:Set(4)
seq-s:Set(1, 111, 11)--seq-v:1111---seq-rs:Set(1, 1111, 111, 11)
seq-s:Set(2, 22)--seq-v:222---seq-rs:Set(222, 2, 22)

(a3,Set(3))
(a4,Set(4))
(a1,Set(1, 1111, 111, 11))
(a2,Set(222, 2, 22))
分区是1的情况
首先判断当前key是否存在，如果存在，那么执行seq代码，将新值追加到已经存在的set中。如果不存在 调用zero的代码生成一个新的set
这里没有执行comb方法，因为我们只有一个分区。

RDD分区为2

根据[11]进行初始化
根据[22]进行初始化
seq-s:Set()--seq-v:11---seq-rs:Set(11)
seq-s:Set()--seq-v:22---seq-rs:Set(22)
根据[3]进行初始化
seq-s:Set(11)--seq-v:1---seq-rs:Set(1, 11)
seq-s:Set()--seq-v:3---seq-rs:Set(3)
seq-s:Set(1, 11)--seq-v:111---seq-rs:Set(1, 111, 11)
根据[4]进行初始化
seq-s:Set()--seq-v:4---seq-rs:Set(4)
根据[2]进行初始化
seq-s:Set()--seq-v:2---seq-rs:Set(2)
根据[1111]进行初始化
seq-s:Set()--seq-v:1111---seq-rs:Set(1111)
seq-s:Set(22)--seq-v:222---seq-rs:Set(222, 22)

comb-s:Set(2)--comb-v:Set(222, 22)---comb-rs:Set(222, 2, 22)
comb-s:Set(1, 111, 11)--comb-v:Set(1111)---comb-rs:Set(1, 1111, 111, 11)

(a3,Set(3))
(a1,Set(1, 1111, 111, 11))
(a4,Set(4))
(a2,Set(222, 2, 22))

“2”和“22”被初始化了两次，说明“a2"被初始化了2次，因为“a2"的数据划分到2个分区中，导致被初始化两次。