键值对 RDD 是 Spark 中许多操作所需要的常见数据类型
键值对 RDD 通常用来进行聚合计算。我们一般要先通过一些初始** ETL(抽取、转化、装载)**操作来将数据转化为键值对形式。
Spark 为包含键值对类型的 RDD 提供了一些专有的操作。
1.创建Pair RDD
1 2 3 4 5 6 7 8 9 10 11 val input = sc.parallelize(List(1, 2, 3, 4)) val pairs = input.map(x => (x+1, x)) for (pair <- pairs){ println(pair) } //输出 (2,1) (3,2) (4,3) (5,4)
2.Pair RDD的转化操作
Pair RDD 可以使用所有标准 RDD 上的可用的转化操作。
Pair RDD也支持RDD所支持的函数
1 2 pairs.filter{case (key, value) => value.length < 20}
3.聚合操作
类似fold() 、 combine() 、 reduce() 等行动操作，这些操作返回 RDD,因此它们是转化操作而不是行动操作。
reduceByKey()函数，接收一个函数，并使用这个函数对值进行合并
1 2 val wordCount = textFile.flatMap(line => line.split(" ")).map(word => (word, 1)).reduceByKey((a, b) => a + b) #切分成单词，转换成键值对并计数
或者
1 2 input.flatMap(x => x.split(" ")).countByValue()
foldByKey()函数，也是接收一个函数，并使用这个函数对值进行合并，提供初始值
1 2 rdd.mapValues(x => (x, 1)).reduceByKey((x, y) => (x._1 + y._1, x._2 + y._2))
countByValue()函数
1 2 3 4 5 val textFile = sc.textFile(inputFile) val result1 = textFile.flatMap(x => x.split(" ")).countByValue() println(result1)<br /> 输出：Map(cc -> 3, aa -> 6, bb -> 3, ee -> 3, ff -> 2, hh -> 1, dd -> 1, gg -> 1)
combineByKey()函数，使用 combineByKey() 求每个键对应的平均值
1 2 3 4 5 6 7 8 9 10 11 12 13 14 val data = Seq(("a", 3), ("b", 4), ("a", 1)) val input2 = sc.parallelize(data) //使用 combineByKey() 求每个键对应的平均值 val result2 = input2.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) } result2.collectAsMap().map(println(_)) //输出 (b,4.0) (a,2.0)
4.数据分组
groupByKey()函数，将数据根据键进行分组
1 2 3 4 5 6 7 8 9 10 val data = Seq(("a", 3), ("b", 4), ("a", 1)) val input2 = sc.parallelize(data) val result3 = input2.groupByKey() for (result <- result3) println(result) //输出： (a,CompactBuffer(3, 1)) (b,CompactBuffer(4))
5.连接
join操作符
1 2 3 4 5 6 7 8 9 10 val input = sc.parallelize(List(1, 2, 3, 4)) val pairs1 = input.map(x => (x+1, x)) //输出(2,1),(3,2),(4,3),(5,4) val pairs2 = input.map(x => (x+1, 1)) //输出(2,1),(3,1),(4,1),(5,1) for(pair <- pairs1.join(pairs2)){ println(pair) } //输出(4,(3,1)),(3,(2,1)),(5,(4,1)),(2,(1,1))
有时，我们不希望结果中的键必须在两个 RDD 中都存在，join的部分可以不存在
leftOuterJoin(other) 和 rightOuterJoin(other) 都会根据键连接两个 RDD，但是允许结果中存在其中的一个 pair RDD 所 缺失的键。
leftOuterJoin(other)结果
1 2 3 4 5 (4,(3,Some(1))) (3,(2,Some(1))) (5,(4,Some(1))) (2,(1,Some(1)))
** rightOuterJoin(other)结果**
1 2 3 4 5 (4,(Some(3),1)) (3,(Some(2),1)) (5,(Some(4),1)) (2,(Some(1),1))
6.数据排序
在 Scala 中以字符串顺序对整数进行自定义排序
1 2 3 4 5 6 7 8 val input = sc.parallelize(List(1, 2, 3, 4)) val pairs1 = input.map(x => (x + 1, x)) implicit val sortIntegersByString = new Ordering[Int] { override def compare(a: Int, b: Int) = b.toString.compare(a.toString) } for(pair <- pairs1.sortByKey()) println(pair) <br />//输出<br />(5,4)<br />(4,3)<br />(3,2)<br />(2,1)
