RDD算子-keyvalue型数据

转换算子

//只对V进行操作
mapValuesU: RDD[(K, U)]
//传进一个分区器，按照分区器进行重新分区
partitionBy(partitioner: Partitioner): RDD[(K, V)]

//把相同key的value聚合到一个seq中
groupByKey(partitioner: Partitioner): RDD[(K, Iterable[V])]
groupByKey(numPartitions: Int): RDD[(K, Iterable[V])]
//按照key排序
sortByKey(ascending: Boolean = true, numPartitions: Int = self.partitions.length): RDD[(K, V)]

//按照相同key对value进行聚合 有combine
reduceByKey(func: (V, V) => V, numPartitions: Int): RDD[(K, V)]
reduceByKey(partitioner: Partitioner, func: (V, V) => V): RDD[(K, V)]
//(每个key的初始值，分区内、间的相同key计算规则)
foldByKey
//(每个key的初始值，分区内相同key计算规则，分区间相同key计算规则)（iPhone，1）（iPhone，1）>>>（iPhone，2）
aggregateByKey
//把V转换结构 转换结构后分区内和分区间操作（李磊，40分）（李磊，30分）>>> (李磊，（30+40，1+1）)
combineByKey

join：inner join两者都有的key才聚合V 把相同key的数据聚合到一个seq
cogroup:full join两者所有K都聚合V 把相同key的数据聚合到一个seq

reduceByKey: 相同 key 的第一个数据不进行任何计算，分区内和分区间计算规则相同
FoldByKey: 相同 key 的第一个数据和初始值进行分区内计算，分区内和分区间计算规则相同
AggregateByKey:相同 key 的第一个数据和初始值进行分区内计算，分区内和分区间计算规则可以不相同
CombineByKey:当计算时，发现数据结构不满足要求时，可以让第一个数据转换结构。分区内和分区间计算规则不相同。

reduceByKey

aggregateByKey

foldByKey

combineByKey

参数描述：
（1）createCombiner: combineByKey() 会遍历分区中的所有key，如果这是一个新的key,combineByKey()会使用一个叫作createCombiner()的函数来创建那个键对应的累加器的初始值.如果已经遇到过这个key 则不作处理
（2）mergeValue: 如果这是一个在处理当前分区之前已经遇到的键，它会使用mergeValue()方法将该键的累加器对应的当前值与这个新的值进行合并
（3）mergeCombiners: 由于每个分区都是独立处理的， 因此对于同一个键可以有多个累加器。如果有两个或者更多的分区都有对应同一个键的累加器， 就需要使用用户提供的 mergeCombiners() 方法将各个分区的结果进行合并。

举栗子：
创建一个pairRDD，根据key计算每种key的均值。（先计算每个key出现的次数以及可以对应值的总和，再相除得到结果）
val input = sc.parallelize(Array((“a”, 88), (“b”, 95), (“a”, 91), (“b”, 93), (“a”, 95), (“b”, 98)),2)
val combine = input.combineByKey((_,1),
// acc是第一步之后的key value格式：（88，1） v是不做处理的value：91
(acc:(Int,Int),v)=>(acc._1+v,acc._2+1),
// 第二步之后的key value
(acc1:(Int,Int),acc2:(Int,Int))=>(acc1._1+acc2._1,acc1._2+acc2._2))
combine.collect >>>Array((b,(286,3)), (a,(274,3)))

val result = combine.map{case (key,value) => (key,value._1/value._2.toDouble)}
result.collect() >>>Array((b,95.33333333333333), (a,91.33333333333333))