Spark笔记8-键值对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)

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)

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)

# 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)

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

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)