1垒探、combineByKey
作用： 將RDD[(K,V)] => RDD[(K,C)] 表示V的類型可以轉(zhuǎn)成C兩者可以不同類型系宫。
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)]
第一個(gè)函數(shù)和第二個(gè)函數(shù)默認(rèn)使用的是HashPartitioner苫幢、serialize為null歇僧。
說(shuō)明：
1）createCombiner:在遍歷RDD的數(shù)據(jù)集合過(guò)程中图张，對(duì)于遍歷到的(k,v)，如果combineByKey第一次遇到值為k的Key（類型K）诈悍，那么將對(duì)這個(gè)(k,v)調(diào)用 createCombiner函數(shù)祸轮，它的作用是將v轉(zhuǎn)換為c(類型是C，聚合對(duì)象的類型侥钳，c作為局和對(duì)象的初始值)适袜。
2）mergeValue： ? ?在遍歷RDD的數(shù)據(jù)集合過(guò)程中，對(duì)于遍歷到的(k,v)舷夺，如果combineByKey不是第一次(或者第二次苦酱，第三次…)遇到值為k的Key（類型K），那么將對(duì)這個(gè) (k,v)調(diào)用mergeValue函數(shù)给猾，它的作用是將v累加到聚合對(duì)象（類型C）中疫萤，mergeValue的類型是(C,V)=>C,參數(shù)中的C遍歷到此處的聚合對(duì)象，然后對(duì)v 進(jìn)行聚合得到新的聚合對(duì)象值敢伸。
3）mergeCombiners：因?yàn)閏ombineByKey是在分布式環(huán)境下執(zhí)行扯饶，RDD的每個(gè)分區(qū)單獨(dú)進(jìn)行combineByKey操作，最后需要對(duì)各個(gè)分區(qū)的結(jié)果進(jìn)行最后的聚合，它的函數(shù)類型是(C,C)=>C尾序，每個(gè)參數(shù)是分區(qū)聚合得到的聚合對(duì)象
例子：
scala> val data = sc.parallelize(List(("1","3"),("1","2"),("1","5"),("2","3")))
scala> val natPairRdd = data.combineByKey(List(_), (c: List[String], v: String) => v::c, (c1: List[String], c2: List[String]) => c1 ::: c2)
scala> natPairRdd.collect
res0: Array[(String, List[String])] = Array((1,List(3, 2, 5)), (2,List(3)))
執(zhí)行的邏輯示意圖：
