Spark学习之四-键值对操作

1.动机

pair RDD:Spark为包含对键值对类型的RDD提供一些专有的操作。

2.创建pair RDD

使用第一个单词创建pair RDD

val pair = lines.map(x=>(x.split(" ")(0),x))

3.转换操作

reduceByKey():聚合相同的key,

实战：

val map_arr = sc.parallelize(List(1,3,5,5,3)) //从集合中生成RDD

val map1 = map_arr.map(x => (x,x+2)) //使用map生成pair RDD

res2: Array[(Int, Int)] = Array((1,3), (3,5), (5,7), (5,7), (3,5))

val sum = map1.reduceByKey((x,y)=>(x+y)) //使用reduceByKey()将key相同的进行聚合

val sum.collect();

结果

res4: Array[(Int, Int)] = Array((1,3), (3,10), (5,14))

groupByKey():对相同键的进行分组

val group_map = map1.groupByKey()

res5: Array[(Int, Iterable[Int])] = Array((1,CompactBuffer(3)), (3,CompactBuffer(5, 5)), (5,CompactBuffer(7, 7)))

mapValues():对pair RDD的每个值进行应用

val mapV = map1.mapValues(x=>(x+10))

res6: Array[(Int, Int)] = Array((1,13), (3,15), (5,17), (5,17), (3,15))

flatMapValues():对pair RDD中的每个值进行操作，

val ss=map1.flatMapValues(x=>(x to 5))；对每个值进行从本身到5的迭代，

res8: Array[(Int, Int)] = Array((1,3), (3,5), (5,7), (5,7), (3,5))

keys:返回pairRDD中所有的key，不需要括号

map1.keys;

values:返回pairRDD中所有的value,不需要括号

map1.values

sortByKey(）:返回一个按键进行排序的RDD

MAP1.sortByKey()

两个RDD 进行操作

rdd1.substractByKey(rdd2):在rdd中删除与rdd2中相同key的元素

rdd1.join(rdd2):内链接

生成二元组的技巧：
①读取文件到rdd中 val lines = sc.textFile("11.txt")

文件内容类似这样

②生成集合 val record_flat=lines.flatMap((x=>xsplit(" ")))
③val record_flat=lines.flatMap((x=>x.split(" ")))
④val map_record=record_flat.map(x=>(1,x))
结果是这样

对pairRDD的值进行筛选

val spy_record=map_line.filter{case(key,value)=>value.length>4} //筛选出值得长度大于4的

聚合操作

解析：使用mapValues对所有的value进行操作，生成（x，1(用来计数的)）二元组，然后根据key进行分区求和，包括，值的求和计数的叠加

rdd.mapValues(x=>(x,1)).reduceByKey((x,y)=>(x._1+y._1,x._2+y._2))


解析 ：单词计数

val lines=sc.textFile("11.txt")

val flat_map=lines.flatMap(x=>x.split(" "))到了这一步可以直接使用.countByValue()简介计算出结果

val map_line=flat_map.map(x=>(x,1))

val count_word=map_line.reduceByKey((x,y)=>(x+y))

combineByKey():求每个键对应的平均值

val result=input.combineByKey(

(v)=>(v,1),生成二元组

(acc:(int,int),v)=>(acc._1+v,acc._2+1),单个分区进行计数和求和

(acc1:(int,int),acc2:(int,int))=>(acc1._1+acc2._1,acc1._2,acc2._2))对分区的值叠加

.map{case (key,value)=>(key,value._1/value._2.toFloat)}

result.coolectAsMap().map(println(_))

并行度调优
每个函数最后一位都用来指定分区数的，没有指定就是用默认的分区数
input.reduceByKey((x,y)=>(x+y),10)10 个分区

数据分组
groupByKey

链接

pair RDD的行动操作
countByKey() ;对每个键对应的元素分别计数
lookup(key):返回指定键对应的所有值