RDD算子介绍

1. RDD算子

RDD算子也叫RDD方法，主要分为两大类：转换和行动。转换，即一个RDD转换为另一个RDD，是功能的转换与补充，比如map，flatMap。行动，则是触发任务的执行，比如collect。所谓算子（Operator），就是通过操作改变问题的状态（来源于认知心理学）。RDD算子有Value类型，双Value类型和Key-Value类型。

2. map

val rdd : RDD[Int] = sc.makeRDD(List(1, 2, 3, 4))
val mapRDD : RDD[Int] = rdd.map(num=>num*2)
mapRDD.collect().foreach(println)

val rdd : RDD[String] = sc.textFile("data")
val mapRDD : RDD[String] = rdd.map(line => {
    val datas = line.split(" ")
    datas(3)
})
mapRDD.collect().foreach(println)

为观察map阶段的分区并行计算过程，添加如下打印

val rdd : RDD[Int] = sc.makeRDD(List(1, 2, 3, 4))
val mapRDD1 : RDD[Int] = rdd.map(num => {
    println(">>>>>>>>")
    num
})
val mapRDD2 : RDD[Int] = rdd.map(num => {
    println("######")
    num
})
mapRDD2.collect().foreach(println)

结果如下：

 

看不出什么规律，改为1个分区：

val rdd : RDD[Int] = sc.makeRDD(List(1, 2, 3, 4), 1)
val mapRDD1 : RDD[Int] = rdd.map(num => {
    println(">>>>>>>>")
    num
})
val mapRDD2 : RDD[Int] = rdd.map(num => {
    println("######")
    num
})
mapRDD2.collect().foreach(println)

结果如下：

 

所以，RDD的计算对于分区内的数据是一个个执行的，即分区内数据的执行是有序的，但是分区间的数据执行是无序的。

3. mapPartitions

上述的map算子对于分区内的数据是一个个依次进行操作，可能存在性能问题，而mapPartitions算子是对于整个分区的数据整体进行操作，但是可能会占用大量空间（以空间换时间）。mapPartitions的参数是iter=>iter。

val rdd : RDD[Int] = sc.makeRDD(List(1, 2, 3, 4), 2)
val mapRDD : RDD[Int] = rdd.mapPartitions(iter => {
    println(">>>>>>>>")
    iter.map(_*2)
})
mapRDD.collect().foreach(println)

结果如下：

 

因为只有两个分区，所以打印两次">>>>>>>>"。使用mapPartitions获取每个分区的最大值：

val rdd : RDD[Int] = sc.makeRDD(List(1, 2, 3, 4), 2)
val mapRDD : RDD[Int] = rdd.mapPartitions(iter => {
    List(iter.max).iterator
})
mapRDD.collect().foreach(println)

这个功能是map算子所实现不了的，因为map算子并不能感知数据来源于分区，而mapPartitions可以以分区为单位进行数据处理（批处理操作）。

4. mapPatitionsWithIndex

mapPartitions虽然以分区为单位进行数据批处理，但是其实也感知不到分区是哪个分区，在一些需要知道分区号的场景下，需要用到mapPatitionsWithIndex。

val rdd : RDD[Int] = sc.makeRDD(List(1, 2, 3, 4), 2)
val mapRDD : RDD[Int] = rdd.mapPartitionsWithIndex((index, iter) => {
    if (index == 1) {
        iter
    } else {
        Nil.iterator
    }
})
mapRDD.collect().foreach(println)

上述代码实现了保留第二个（索引为1）分区，结果如下：

 

val rdd : RDD[Int] = sc.makeRDD(List(1, 2, 3, 4))
val mapRDD : RDD[Int] = rdd.mapPartitionsWithIndex((index, iter) => {
    iter.map(num => (index, num))
})
mapRDD.collect().foreach(println)

上述代码实现了查看每个数据在哪个分区，结果如下：

5. flatMap

flatMap做扁平化映射

val rdd : RDD[List[Int]] = sc.makeRDD(List(List(1, 2), List(3, 4)))
val mapRDD : RDD[Int] = rdd.flatMap(list => list)
mapRDD.collect().foreach(println)

val rdd : RDD[String] = sc.makeRDD(List("Hello Spark", "Hello Scala"))
val mapRDD : RDD[String] = rdd.flatMap(s => s.split(" "))
mapRDD.collect().foreach(println)

结果如下：

 

将List(List(1,2), 3, List(4,5))进行扁平化操作（使用模式匹配）：

val rdd : RDD[List[Int]] = sc.makeRDD(List(List(1, 2), 3, List(4, 5)))
val mapRDD : RDD[Int] = rdd.flatMap(data => {
    data match {
        case list:List[] => list
        case num => List(num)
    }
})
mapRDD.collect().foreach(println)

6. glom

glom操作有点类似于flatMap的逆操作，将分区内的数据转换为相同类型的内存数组，分区不变。

val rdd : RDD[Int] = sc.makeRDD(List(1, 2, 3, 4), 2)
val glomRDD : RDD[Array[Int]] = rdd.glom()
glomRDD.collect().foreach(data=>data.mkString(","))

结果如下：

 

求各分区最大值之和：

val rdd : RDD[Int] = sc.makeRDD(List(1, 2, 3, 4), 2)
val glomRDD : RDD[Array[Int]] = rdd.glom()
val maxRDD : RDD[Int] = glomRDD.map(array => array.max)
println(maxRDD.collect().sum))

7. groupBy

按照指定的key进行分组

val rdd : RDD[Int] = sc.makeRDD(List(1, 2, 3, 4), 2)
val groupRDD : RDD[(Int, Iterable[Int])] = rdd.groupBy(num => num % 2)
groupRDD.collect().foreach(println)

结果如下：

 

按照首字母分组

val rdd : RDD[String] = sc.makeRDD(List("Hello", "Spark", "Scala", "Hadoop"), 2)
val groupRDD = rdd.groupBy(s => s.charAt(0))
groupRDD.collect().foreach(println)

结果如下：

 

分组的过程可能会打乱数据，即数据可能会重新组合，原分区的数据被分到另一个分区了，即shuffle过程。极限情况下，数据可能被分到一个分区中。一个组的数据在一个分区中，但是一个分区不一定只有一个组。

8. filter

过滤偶数

val rdd : RDD[Int] = sc.makeRDD(List(1, 2, 3, 4), 2)
val filterRDD : RDD[Int] = rdd.filter(num % 2 == 0)
filterRDD.collect().foreach(println)

按照指定规则进行数据过滤，分区不变，过滤后，不同分区内的数据可能不均衡，即数据倾斜。 

过滤指定日期的数据：

val rdd : RDD[String] = sc.textFile("data")
val filterRDD : RDD[String] = rdd.filter(line => {
    val datas = line.split(" ")
    datas(3).startWith("17/05/2015")
})
filterRDD.collect().foreach(println)

9. sample

采样/抽取数据，用的一般不多，其中一个用途可能是解决数据倾斜问题。sample算子主要有三个参数，第一个是抽取的数放不放回去，第二个参数是概率，如果抽取不放回，则表示每个数被抽取的概率，如果抽取放回，则表示某个数可能的抽取次数（可能的次数而已），第三个参数是随机数算法种子（一般可不填，如果填了，可能会导致抽取结果固定）。

val rdd : RDD[Int] = sc.makeRDD(List(1, 2, 3, 4, 5, 6, 7, 8, 9, 10))
val sampleRDD : RDD[Int] = rdd.sample(false, 0.4, 1)
println(sampleRDD.collect().mkstring(","))

 结果如下：

多运行几次，发现结果不变，因为随机数算法种子固定了，如果不传，则默认使用系统时间（变化的）。

val rdd : RDD[Int] = sc.makeRDD(List(1, 2, 3, 4, 5, 6, 7, 8, 9, 10))
val sampleRDD : RDD[Int] = rdd.sample(false, 0.4)
println(sampleRDD.collect().mkstring(","))

此时结果就不固定，结果都不一定为4个数。

根据源码，如果抽取不放回，抽取算法为伯努利分布，如果抽取放回，则为泊松分布。

如果抽取放回，

val rdd : RDD[Int] = sc.makeRDD(List(1, 2, 3, 4, 5, 6, 7, 8, 9, 10))
val sampleRDD : RDD[Int] = rdd.sample(true, 2)
println(sampleRDD.collect().mkstring(","))

 结果如下：

10. distinct

distinct算子用于去重

val rdd : RDD[Int] = sc.makeRDD(List(1, 2, 3, 4, 1, 2, 3, 4))
val distinctRDD : RDD[Int] = rdd.distinct()
distinctRDD.collect().foreach(println)