Spark--RDD中的转换算子

1、算子的简单介绍

Transformation(转换)算子:根据数据集创建一个新的数据集,计算后返回一个新RDD,例如一个rdd进行map操作后生了一个新的rdd。

Action(动作)算子:对rdd结果计算后返回一个数值value给驱动程序(driver),例如collect算子将数据集的所有元素收集完成返回给驱动程序。

控制算子:对数据集进行特殊操作,例如cache算子将对于重复使用的算子,进行cache做缓存使用,数据只保存在内存中,性能提升。

懒执行:Spark中转化算子和控制算子是懒执行的,需要Action算子触发才能执行。

懒执行就是延迟计算的意思,就像是创建了一个视图,他并不是把查询好的数据放入视图了,而是当你需要这些数据时,查看视图时,他才执行定义视图时候的SQL语句。

注意:

Driver即运行Application的main()函数并且创建SparkContext。

Application用户编写的Spark应用程序。

SparkContext整个应用的上下文、控制应用的生命周期。

job即在每一个application中,有几个action,就会产生几个job。

2、算子的使用

·map 算子

import org.apache.spark.{SparkConf, SparkContext}
object MapExample {def main(args: Array[String]): Unit = {val conf = new SparkConf().setAppName("MapExample").setMaster("local[*]")val sc = new SparkContext(conf)val rdd = sc.parallelize(Seq(1, 2, 3, 4))val newRdd = rdd.map(x => x * 2)newRdd.collect().foreach(println)sc.stop()}
}

import org.apache.spark.{SparkConf, SparkContext}
object FilterExample {def main(args: Array[String]): Unit = {val conf = new SparkConf().setAppName("FilterExample").setMaster("local[*]")val sc = new SparkContext(conf)val rdd = sc.parallelize(Seq(1, 2, 3, 4))val newRdd = rdd.filter(x => x % 2 == 0)newRdd.collect().foreach(println)sc.stop()
}}

import org.apache.spark.{SparkConf, SparkContext}
object FlatMapExample {def main(args: Array[String]): Unit = {val conf = new SparkConf().setAppName("FlatMapExample").setMaster("local[*]")val sc = new SparkContext(conf)val rdd = sc.parallelize(Seq("hello world", "spark is great"))val newRdd = rdd.flatMap(x => x.split(" "))newRdd.collect().foreach(println)sc.stop()}}

import org.apache.spark.{SparkConf, SparkContext}
object ReduceByKeyExample {def main(args: Array[String]): Unit = {// 创建 SparkConf 对象val conf = new SparkConf().setAppName("ReduceByKeyExample").setMaster("local[*]")// 创建 SparkContext 对象val sc = new SparkContext(conf)// 创建一个包含单词的 RDDval words = sc.parallelize(List("apple", "banana", "apple", "cherry", "banana", "apple"))// 将每个单词映射为 (单词, 1) 的键值对val wordPairs = words.map(word => (word, 1))// 使用 reduceByKey 计算每个单词的出现次数val wordCounts = wordPairs.reduceByKey(_ + _)// 输出结果wordCounts.collect().foreach(println)// 停止 SparkContextsc.stop()}
}

·Transformation算子   

val conf = new SparkConf().setAppName("WordCount").setMaster("local")val sc = new SparkContext(conf)val pairRdd = sc.parallelize(List((1,1), (5,10), (5,9), (2,4), (3,5), (3,6),(4,7), (4,8),(2,3), (1,2)),4)//map(函数)，一一映射，分区数量不变，有多少条数据，就被会运行多少次。val value1: RDD[(Int, Int)] = pairRdd.map(x => (x._1, x._2 + 1))//定义一个迭代函数val f1 = (iter:Iterator[(Int, Int)]) => iter.map(x=>(x._1,x._2*2))//mapPartitions函数可以认为是Map的变种，可以对分区进行并行处理，两者的区别是调用的颗粒度不一样，map的输入函数是应用于RDD的每个元素，而mapPartition的输入函数是应用于RDD的每个分区。val value2: RDD[(Int, Int)] = pairRdd.mapPartitions(f1)//定义一个迭代函数val f2 = (index:Int,iter:Iterator[(Int, Int)]) => iter.map((index,_))//mapPartitionsWithIndex函数类似于mapPartitions，但func带有一个整数参数表示分片的索引值，因此在类型为T的RDD上运行时，func的函数类型必须是(Int, Interator[T]) => Iterator[U]val value3: RDD[(Int, (Int, Int))] = pairRdd.mapPartitionsWithIndex(f2)//filter函数,返回一个新的RDD，该RDD由经过func函数计算后返回值为true的输入元素组成,fitler并不会改变分区的数量，之前有几个，现在仍然有几个分区。val value4: RDD[(Int, Int)] = pairRdd.filter(x => x._2 % x._1 == 0)//flatMap函数: map之后，再flatten。每一个输入元素可以被映射为0或多个输出元素（所以func应该返回一个序列，而不是单一元素）val value5: RDD[Char] = pairRdd.flatMap(x => x._1.toString + x._2.toString)//groupByKey函数:在一个(K,V)的RDD上调用，返回一个(K, Iterator[V])的RDD,可重新指定分区个数val value6: RDD[(Int, Iterable[Int])] = pairRdd.groupByKey(1)//groupBy函数与groupByKey类似,底层调用groupByKey,返回一个(K, Iterator[V])的RDD,可重新指定分区个数val value7: RDD[(Int, Iterable[(Int, Int)])] = pairRdd.groupBy(_._1, 1)//groupBy后要搭配MapValues()使用//reduceByKey函数:在一个(K,V)的RDD上调用，返回一个(K,V)的RDD，使用指定的reduce函数，将相同key的值聚合到一起，与groupByKey类似，reduce任务的个数可以通过第二个可选的参数来设置val value8: RDD[(Int, Int)] = pairRdd.reduceByKey(_ + _, 1)//aggregateByKey 聚合类算子：初始化在每一个分区聚合中参数运行，但是在全局聚合中，不参与,类似于reduceByKey,但多个初始值val value11: RDD[(Int, Int)] = pairRdd.aggregateByKey(100)(_ + _, _ + _)//sortByKey函数:在一个(K,V)的RDD上调用，K必须实现Ordered接口，返回一个按照key进行排序的(K,V)的RDD,是全局排序,可指定分区数量//可设置排序规则,默认是从小到大排序，是true， 如果想从大到小排，设置为falseval value12: RDD[(Int, Int)] = pairRdd.sortByKey(false,2)//sortBy函数:与sortByKey类似，但是更灵活,可设置排序的字段val value13: RDD[(Int, Int)] = pairRdd.sortBy(_._1, false, 2)//union函数:对源RDD和参数RDD求并集后返回一个新的RDDval value14: RDD[(Int, Int)] = value1.union(value2)//intersection函数:对源RDD和参数RDD求交集后返回一个新的RDDval value15: RDD[(Int, Int)] = value1.intersection(value2)//subtract函数:对源RDD和参数RDD求差集后返回一个新的RDDval value16: RDD[(Int, Int)] = value1.subtract(value2)//join函数:对源RDD和参数RDD进行join返回一个新的RDDval value17: RDD[(Int, (Int, Int))] = value1.join(value2)//cogroup函数:对源RDD和参数RDD进行全外关联返回一个新的RDD,相当于SQL中的全外关联full outer join，返回左右RDD中的记录，关联不上的为空。val value18: RDD[(Int, (Iterable[Int], Iterable[Int]))] = value1.cogroup(value2)//cartesian函数:对源RDD和参数RDD进行笛卡尔积返回一个新的RDDval value19: RDD[((Int, Int), (Int, Int))] = value1.cartesian(value2)//coalesce:对RDD重新进行分区val value20: RDD[((Int, Int), (Int, Int))] = value19.coalesce(10)//repartition:类似于coalesce方法，底层就是用的coalesce方法，对RDD重新进行分区val value21: RDD[((Int, Int), (Int, Int))] = value20.repartition(5)//去重：distinct(分区数量)。val value22: RDD[((Int, Int), (Int, Int))] = value19.distinct(12)//等价于val value23: RDD[((Int, Int), (Int, Int))] =value19.map(x=>(x,null)).reduceByKey((x,y) => x).map(_._1)

·ACtion算子   

//reduce函数与reduceByKey类似,返回一个结果的RDD,+ ,++ 取决rdd的元素类型,string类型使用++val value9: Int = pairRdd.map(_._2).reduce(_ + _)//aggregate 聚合类算子：存在两次聚合。局部聚合和全局聚合。//先局部计算(100+分区一的内容) (100+分区二的内容)+...//再全局计算100+(100+分区一的内容)+ (100+分区二的内容)+...val value10: Int = pairRdd.map(_._2).aggregate(100)(_ + _, _ + _)//collect 在驱动程序中，以数组的形式返回数据集的所有元素val tuples: Array[((Int, Int), (Int, Int))] = value21.collect()//foreach 在数据集的每一个元素上，运行函数funcval unit: Unit = value11.foreach(println(_))//foreach 迭代的是每一个分区的数据//如果我们需要去获取mysql的连接，RDD中有10000 条数据，有10个分区。//foreach：获取10000次连接，性能低下。//foreapartition: 只需要获取10次连接，每一个分区中的数据，共用一个连接。 性能优越。value21.foreachPartition(println(_))//saveAsTextFile 将数据集的元素以textfile的形式保存到HDFS文件系统或者其他支持的文件系统，对于每个元素，Spark将会调用toString方法，将它装换为文件中的文本value21.saveAsTextFile("路径")//count() 返回RDD的元素个数val tuples1: Array[((Int, Int), (Int, Int))] = value19.collect()//first() 返回RDD的第一个元素（类似于take(1)）val tuple: ((Int, Int), (Int, Int)) = value19.first()//take(n) 返回一个由数据集的前n个元素组成的数组val tuples2: Array[((Int, Int), (Int, Int))] = value19.take(10)

注意：

sortBy 是按照 RangePartitioner作为分区器groupByKey 和 reduceByKey 中优先使用reduceByKey,因为reduceByKey会有一个局部的聚合,性能更好。

ByKey的都是转换算子,没有Key的都是action类算子。aggregateByKey和reduceByKey是转化算子,aggregate和reduce是执行算子

2.3、控制算子

cache 对于重复使用的算子,进行cache做缓存使用,数据只保存在内存中,性能提升persist 性能提升checkPoint 数据容错,当数据计算的时候,机器挂了,重新追溯到checkPoint的目录下checkPoint是将RDD持久化到磁盘中,还可以切断RDD之间的依赖关系。