【Spark学习笔记】KeyValue对RDDs

创建KeyValue对RDDs：使用map()函数，返回key/value对
例如，包含数行数据的RDD，把每行数据的第一个单词作为keys

val rdd2=rdd.map(line=>(line.split(" ")(0),line))

image.png

val rdd3=sc.parallelize(Array((1,2),(3,4),(3,6)))
(1,2)
(3,4)
(3,6)

val rdd4=rdd3.reduceByKey((x,y)=x+y)
(1,2)
(3,10)

val rdd5=rdd3.groupByKey()
(1,2)
(3,(4,6))

image.png

combineByKey()

(createCombiner,mergeValue,mergeCombiners,partitioner)
最常用的基于key的聚合函数，返回的类型可以与输入类型不一样，许多基于key的聚合函数都用到了它，像groupByKey()

combineByKey()
遍历partition中的元素，元素的key，要么是之前见过的，要么不是。
如果是新元素，使用我们提供的createCombiner()函数
如果是这个partition中已经存在的key，就会使用mergeValue()函数合计每个partition的结果的时候，使用mergeCombiners()函数。

val scores=sc.parallelize(Array(("jake",80.0),("jake",90.0),("jake",85.0),("mike",92.0),("mike",90.0)))
val score2=score.combineByKey(score=>(1,score),(c1:(Int,Double),newScore)=>(c1._1+1,c1._2+newScore),(c1:(Int,Double),c2:(Int,Double)=>(c1._1+c2_1,c1._2+c2._2))
score2.foreach(println)
(jake,(3,255.0))
(mike,(3,267.0))

val average=scores2.map{case(name,(num,score))=>(name,score/num)}
average.foreach(println)
(mike,89.0)
(jake,85.0)