Spark笔记8-键值对RDD

键值对RDD

创建

1. 从数据中进行加载生成键值对RDD

lines = sc.textFile("word.txt")
pairRDD = lines.flatMap(lambda line: line.split(" ")) \ # 将数据先进行分割split，再拍平flat，形成单个的元素
               .map(lambda word:(word, 1))  # 单个元素和1组成元组的形式，键值对RDD
pairRDD.foreach(print)

("hadoop", 1)
("spark", 1)
("hive", 1)

2. 并行集合组成键值对RDD

   list = ["hadoop", "spark", "hive"]
   rdd = sc.parallelize(list)
   pairRDD = rdd.map(lambda word:(word, 1))  # 转成键值对RDD
   pairRDD.foreach(print)

常见转换

reduceByKey(func)

先通过key进行分组，再通过value进行func函数的运用

pairRDD = sc.parallelize([("hadoop",1),("hive",1),("spark", 1), ("spark", 1)])
pairRDD.reduceByKey(lambda a,b: a+b).foreach(print)   # 先通过key进行分组，再通过value进行func函数的运用

("spark",2)
("hive",1)
("hadoop",1)

groupByKey()

不进行聚合运算，只是将相同键的值进行合并成一个列表的形式。

pairRDD = sc.parallelize([("hadoop",1),("hive",1),("spark", 1), ("spark", 1)])
pairRDD.reduceByKey(lambda a,b: a+b).foreach(print)   # 先通过key进行分组，再通过value进行func函数的运用

("spark",(1, 1))  # 不进行计算，只合并成一个元组形式  
("hive",1)
("hadoop",1)

​ reduceByKey可以等价于groupByKey加上map操作：

# 1
words = ["one","two","two", "three", "three", "three"]
wordpairRDD = sc.parallelize(words).map(lambda word:(word,1))
wordCountReduce = wordpairRDD.reduceByKey(lambda a.b:a+b)
wordCountReduce.foreach(print)

# 2
wordcountGroup = wordpairRDD.groupByKey().map(lambda t:(t[0], sum(t[1])))
wordCountGroup.foreach(print)

("one",1)
("two",2)
("three",3)

keys()/vlaues()

只取出key/value的值

pairRDD = sc.parallelize([("hadoop",1),("hive",1),("spark", 1), ("spark", 1)])
pairRDD.keys().foreach(print)

sortBykey()

对key进行排序，默认是升序False

mapValues(func)

key保持不变，值根据func函数进行改变

pairRDD = sc.parallelize([("hadoop",1),("hive",1),("spark", 1), ("spark", 1)])
pairRDD1 = pairRDD.mapValues(lambda x:x+1)  # 将每个值进行加1操作
pairRDD1.foreach(print)

("hadoop",2)
("hive",2)
("spark",2)
("spark",2)

join

如果两个RDD的key相同，将它们的value进行合并，形成一个元组的形式。

综合demo

• mapValues：只对value进行操作，而且没有聚合操作
• reduceByKey(func)：先分组，再对value进行函数func的聚合操作