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[] => listcase 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)