Spark学习笔记——键值对操作

键值对 RDD是 Spark 中许多操作所需要的常见数据类型

键值对 RDD 通常用来进行聚合计算。我们一般要先通过一些初始** ETL(抽取、转化、装载)**操作来将数据转化为键值对形式。

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

 

1.创建Pair RDD

    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所支持的函数

pairs.filter{case (key, value) => value.length < 20}

 

3.聚合操作

类似fold() 、 combine() 、 reduce() 等行动操作，这些操作返回 RDD,因此它们是转化操作而不是行动操作。

reduceByKey()函数，接收一个函数，并使用这个函数对值进行合并

val wordCount = textFile.flatMap(line => line.split(" ")).map(word => (word, 1)).reduceByKey((a, b) => a + b)　　#切分成单词，转换成键值对并计数

或者

input.flatMap(x => x.split(" ")).countByValue()

foldByKey()函数，也是接收一个函数，并使用这个函数对值进行合并，提供初始值

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

 

countByValue()函数

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() 求每个键对应的平均值

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()函数，将数据根据键进行分组

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操作符

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)结果

(4,(3,Some(1)))
(3,(2,Some(1)))
(5,(4,Some(1)))
(2,(1,Some(1)))

** rightOuterJoin(other)结果**

(4,(Some(3),1))
(3,(Some(2),1))
(5,(Some(4),1))
(2,(Some(1),1))

 

6.数据排序

在 Scala 中以字符串顺序对整数进行自定义排序

    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)