combineByKey

def combineByKey[C](createCombiner: (V) => C, mergeValue: (C, V) => C, mergeCombiners: (C, C) => C): RDD[(K, C)]

def combineByKey[C](createCombiner: (V) => C, mergeValue: (C, V) => C, mergeCombiners: (C, C) => C, numPartitions: Int): RDD[(K, C)]

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)]

该函数用于将RDD[K,V]转换成RDD[K,C],这里的V类型和C类型可以相同也可以不同。

其中的参数：

createCombiner：组合器函数，用于将V类型转换成C类型，输入参数为RDD[K,V]中的V,输出为C

mergeValue：合并值函数，将一个C类型和一个V类型值合并成一个C类型，输入参数为(C,V)，输出为C

mergeCombiners：合并组合器函数，用于将两个C类型值合并成一个C类型，输入参数为(C,C)，输出为C

numPartitions：结果RDD分区数，默认保持原有的分区数

partitioner：分区函数,默认为HashPartitioner

mapSideCombine：是否需要在Map端进行combine操作，类似于MapReduce中的combine，默认为true

看下面例子：

def main(args: Array[String]): Unit = {//默认分区12个val sc = new SparkContext(new SparkConf().setMaster("local").setAppName("test").set("spark.default.parallelism", "12"))val rdd1 = sc.makeRDD(Array((1, "A"), (1, "B"), (2, "A"), (2, "D"), (3, "E"), (1, "A")))rdd1.combineByKey((v: String) => List(v),(c: List[String], v: String) => c.::(v),(c1: List[String], c2: List[String]) => c1 ::: c2,2).collect.foreach(println(_))}

16/12/20 15:59:37 INFO DAGScheduler: ResultStage 1 (collect at ShellTest.scala:27) finished in 0.216 s
16/12/20 15:59:37 INFO DAGScheduler: Job 0 finished: collect at ShellTest.scala:27, took 1.356017 s
(2,List(A, D))
(1,List(A, B, A))
(3,List(E))
16/12/20 15:59:37 INFO SparkContext: Invoking stop() from shutdown hook

foldByKey

def foldByKey(zeroValue: V)(func: (V, V) => V): RDD[(K, V)]

def foldByKey(zeroValue: V, numPartitions: Int)(func: (V, V) => V): RDD[(K, V)]

def foldByKey(zeroValue: V, partitioner: Partitioner)(func: (V, V) => V): RDD[(K, V)]

该函数用于RDD[K,V]根据K将V做折叠、合并处理，其中的参数zeroValue表示先根据映射函数将zeroValue应用于V,进行初始化V,再将映射函数应用于初始化后的V.

直接看例子：

1. scala> var rdd1 = sc.makeRDD(Array(("A",0),("A",2),("B",1),("B",2),("C",1)))
2. scala> rdd1.foldByKey(0)(_+_).collect
3. res75: Array[(String, Int)] = Array((A,2), (B,3), (C,1))
4. //将rdd1中每个key对应的V进行累加，注意zeroValue=0,需要先初始化V,映射函数为+操
5. //作，比如("A",0), ("A",2)，先将zeroValue应用于每个V,得到：("A",0+0), ("A",2+0)，即：
6. //("A",0), ("A",2)，再将映射函数应用于初始化后的V，最后得到(A,0+2),即(A,2)

再看：

1. scala> rdd1.foldByKey(2)(_+_).collect
2. res76: Array[(String, Int)] = Array((A,6), (B,7), (C,3))
3. //先将zeroValue=2应用于每个V,得到：("A",0+2), ("A",2+2)，即：("A",2), ("A",4)，再将映射函
4. //数应用于初始化后的V，最后得到：(A,2+4)，即：(A,6)

再看乘法操作：

1. scala> rdd1.foldByKey(0)(_*_).collect
2. res77: Array[(String, Int)] = Array((A,0), (B,0), (C,0))
3. //先将zeroValue=0应用于每个V,注意，这次映射函数为乘法，得到：("A",0*0), ("A",2*0)，
4. //即：("A",0), ("A",0)，再将映射函//数应用于初始化后的V，最后得到：(A,0*0)，即：(A,0)
5. //其他K也一样，最终都得到了V=0
6. scala> rdd1.foldByKey(1)(_*_).collect
7. res78: Array[(String, Int)] = Array((A,0), (B,2), (C,1))
8. //映射函数为乘法时，需要将zeroValue设为1，才能得到我们想要的结果。

在使用foldByKey算子时候，要特别注意映射函数及zeroValue的取值。