一、map算子
将处理的数据源的数据逐条进行映射转换，这里的转换可以是类型的转换，也可以是值的转换。
map的参数为一个函数。在map转换后，分区是不变的。
例子：将每个分区的数据都乘以2
def main(args: Array[String]): Unit = {
val sparkConf = new SparkConf().setMaster("local[*]").setAppName("Operator")
val sc = new SparkContext(sparkConf)
val rdd = sc.makeRDD(List(1, 2, 3, 4),2)
val mapRDD: RDD[Int] = rdd.map(_ * 2)
mapRDD.collect().foreach(println)
sc.stop()
}
二、mapPartitions算子
mapPartitions以分区为单位进行数据转换操作，每个分区是一个迭代器。
缺点是会将整个分区的数据加载到内存进行引用，处理完的数据是不会被释放掉，因为存在对象的引用，在内存较小，数据量较大的场合下，容易出现内存溢出。
例子：取得每个分区的最大值
def main(args: Array[String]): Unit = {
val sparkConf = new SparkConf().setMaster("local[*]").setAppName("Operator")
val sc = new SparkContext(sparkConf)
val rdd = sc.makeRDD(List(1, 2, 3, 4), 2)
val mapRDD: RDD[Int] = rdd.mapPartitions(iter => { List(iter.max).iterator })
mapRDD.collect().foreach(println)
sc.stop()
}
map与mapPartitions的区别：
- 数据处理角度
Map 算子是分区内一个数据一个数据的执行，类似于串行操作。而 mapPartitions 算子
是以分区为单位进行批处理操作。
- 功能的角度
Map 算子主要目的将数据源中的数据进行转换和改变。但是不会减少或增多数据。
MapPartitions 算子需要传递一个迭代器，返回一个迭代器，没有要求的元素的个数保持不变，
所以可以增加或减少数据
- 性能的角度
Map 算子因为类似于串行操作，所以性能比较低，而是 mapPartitions 算子类似于批处
理，所以性能较高。但是 mapPartitions 算子会长时间占用内存，那么这样会导致内存可能
不够用，出现内存溢出的错误。所以在内存有限的情况下，不推荐使用。使用 map 操作。
三、mapPartitionsWithIndex算子
和mapPartitions的功能基本相同，不同点在于在处理时同时可以获取当前分区的索引。
例子：获取第二个分区的数据
def main(args: Array[String]): Unit = {
val sparkConf = new SparkConf().setMaster("local[*]").setAppName("Operator")
val sc = new SparkContext(sparkConf)
val rdd = 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)
sc.stop()
}
四、flatMap算子
flatMap先将处理的数据进行扁平化（flat）后再进行映射（map）处理，所以flatMap算子也称之为扁平映射。
flat时需要传入list对象,后一个list是map阶段映射后需要再传给flat的内容
例子：扁平化列表
def main(args: Array[String]): Unit = {
val sparkConf = new SparkConf().setMaster("local[*]").setAppName("Operator")
val sc = new SparkContext(sparkConf)
val rdd = sc.makeRDD(List(List(1, 2), List(3, 4)), 2)
val flatRDD: RDD[Int] = rdd.flatMap( list => { list } )
flatRDD.collect().foreach(println)
sc.stop()
}
五、glom算子
glom算子可以将每个分区的数据直接转换为相同类型的Array数组进行处理，分区不变
例子：每个分区转化为array数组后取出每个分区的最大值再求和
def main(args: Array[String]): Unit = {
val sparkConf = new SparkConf().setMaster("local[*]").setAppName("Operator")
val sc = new SparkContext(sparkConf)
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)
sc.stop()
}
|