spark算子_Spark常用算子
Spark的算子分类：
从大方向说，Spark算子大致可以分为以下两类：
（1）Transformation变换/转换算子：这种变换并不触发提交作业，这种算子是延迟执行的，也就是说从一个RDD转换生成另一个RDD的转换操作不是马上执行，需要等到有Action操作的时候才会真正触发。
（2）Action行动算子：这类算子会触发SparkContext提交job作业，并将数据输出到Spark系统。
从小方向说，Spark算子大致可以分为以下三类：
（1）Value数据类型的Transformation算子，这种变换并不触发提交作业，针对处理的数据项是Value型的数据
（2）Key-Value 数据类型的Transformation算子，这种变换并不触发提交作业，针对处理的数据项是Key-Value型的数据对
（3）Action算子：这类算子会触发SparkContext提交Job作业
（一）Value数据类型的Transformation算子：
（1）输入分区与输出分区一对一型：
1.map算子
2.flatMap算子
3.mapPartitions算子
4.mapPartitionsWithIndex算子
（2）输入分区与输出分区多对一型
5.union算子
6.cartesian算子
（3）输入分区与输出分区多对多型
7.groupBy算子、groupByKey算子
（4）输出分区是输入分区子集类型
8.filter算子
9.distinct算子
10.subtract算子
11.sample算子
（5）Cache算子
13.cache算子
14.persist算子
（二）Key-Value数据类型的Transformation算子
（1）输入分区与输出分区一对一
15.mapValues算子
（2）对单个RDD或者两个RDD聚集
单个RDD聚集
16.combineByKey算子
17.reduceByKey算子
18.repartition算子
两个RDD聚集
19.cogroup算子
（3）连接
20.join算子
21.leftOutJoin和rightOutJoin算子、fullOuterJoin算子
（三）Action算子
（1）无输出
22.foreach算子
（2）HDFS
23.saveAsTextFile算子
24.saveAsObjectFile算子
（3）Scala集合和数据类型
25.collect算子
26.collectAsMap算子
27.count，countByKey，CountByValue算子
28.take、takeSample算子
29.reduce算子
30.aggregate算子
31.zip、zipWithIndex算子
Transformation：
1.map算子
处理数据是一对一的关系，进入一条数据，出去的还是一条数据。map的输入变换函数应用于RDD中所有的元素，而mapPartitions应用于所有分区。区别于mapPartitions主要在于调度粒度不同。如parallelize（1 to 10 ，3），map函数执行了10次，而mapPartitions函数执行了3次。
val infos: RDD[String] = sc.parallelize(Array[String]("hello spark","hello hdfs","hello HBase"))
val result: RDD[Array[String]] = infos.map(one => {one.split(" ")
})
result.foreach(arr =>{arr.foreach(println)})执行结果：
2.flatMap算子
flatMap是一对多的关系，处理一条数据得到多条结果
将原来 RDD 中的每个元素通过函数 f 转换为新的元素，并将生成的 RDD 的每个集合中的元素合并为一个集合。
val infos: RDD[String] = sc.makeRDD(Array[String]("hello spark","hello hdfs","hello MapReduce"))
val rdd1: RDD[String] = infos.flatMap(one => {one.split(" ")
})
rdd1.foreach(println)3.mapPartitions算子
mapPartitions遍历的是每一个分区中的数据，一个个分区的遍历。获 取 到 每 个 分 区 的 迭 代器，在 函 数 中 通 过 这 个 分 区 整 体 的 迭 代 器 对整 个 分 区 的 元 素 进 行 操 作，相对于map一条条处理数据，性能比较高，可获取返回值。
可以通过函数f(iter) =>iter.filter(_>=3)对分区中所有的数据进行过滤，大于和等于3的数据保留，一个方块代表一个RDD分区，含有1,2,3的分区过滤，只剩下元素3。
4.mapPartitionsWithIndex(function)算子
拿到每个RDD中的分区，以及分区中的数据
val lines: RDD[String] = sc.textFile("./data/words",5)val result: RDD[String] = lines.mapPartitionsWithIndex((index, iter) => {val arr: ArrayBuffer[String] = ArrayBuffer[String]()iter.foreach(one => {
// one.split(" ")arr.append(s"partition = [$index] ,value = $one")})arr.iterator}, true)result.foreach(println)5.union算子
union合并两个RDD，两个RDD必须是同种类型，不一定是K，V格式的RDD
val rdd1: RDD[String] = sc.parallelize(List[String]("zhangsan","lisi","wangwu","maliu"),3)
val rdd2: RDD[String] = sc.parallelize(List[String]("a","b","c","d"),4)
val unionRDD: RDD[String] = rdd1.union(rdd2)
unionRDD.foreach(println)6.cartesian算子
求笛卡尔积，该操作不会执行shuffle操作，但最好别用，容易触发OOM
7.groupBy算子
按照指定的规则，将数据分组
val rdd: RDD[(String, Double)] = sc.parallelize(List[(String,Double)](("zhangsan",66.5),("lisi",33.2),("zhangsan",66.7),("lisi",33.4),("zhangsan",66.8),("wangwu",29.8)))
val result: RDD[(Boolean, Iterable[(String, Double)])] = rdd.groupBy(one => {one._2 > 34
})
result.foreach(println)groupByKey算子
根据key去将相同的key对应的value合并在一起（K,V）=>(K,[V])
val rdd: RDD[(String, Double)] = sc.parallelize(List[(String,Double)](("zhangsan",66.5),("lisi",33.2),("zhangsan",66.7),("lisi",33.4),("zhangsan",66.8),("wangwu",29.8)))
val rdd1: RDD[(String, Iterable[Double])] = rdd.groupByKey()
rdd1.foreach(info=>{val name: String = info._1val value: Iterable[Double] = info._2val list: List[Double] = info._2.toListprintln("name = "+name+",value ="+list)
})8.filter算子
过滤数据，返回true的数据会被留下
val infos: RDD[String] = sc.makeRDD(List[String]("hehe","hahha","zhangsan","lisi","wangwu"))
val result: RDD[String] = infos.filter(one => {!one.equals("zhangsan")
})
result.foreach(println)9.distinct算子
distinct去重，有shuffle产生，内部实际是map+reduceByKey+map实现
val infos: RDD[String] = sc.parallelize(List[String]("a","a","b","a","b","c","c","d"),4)
val result: RDD[String] = infos.distinct()
result.foreach(println)10.subtract算子
取RDD的差集，subtract两个RDD的类型要一致，结果RDD的分区数与subtract算子前面的RDD分区数多的一致。
val rdd1 = sc.parallelize(List[String]("zhangsan","lisi","wangwu"),5)
val rdd2 = sc.parallelize(List[String]("zhangsan","lisi","maliu"),4)
val subtractRDD: RDD[String] = rdd1.subtract(rdd2)
subtractRDD.foreach(println)
println("subtractRDD partition length = "+subtractRDD.getNumPartitions)11.sample算子
sample随机抽样，参数sample（withReplacement:有无放回抽样，fraction:抽样的比例，seed:用于指定的随机数生成器的种子）
有种子和无种子的区别：
有种子是只要针对数据源一样，都是指定相同的参数，那么每次抽样到的数据都是一样的
没有种子是针对同一个数据源，每次抽样都是随机抽样
(12.13)cache算子、persist算子
package core.persistimport org.apache.spark.rdd.RDD
import org.apache.spark.storage.StorageLevel
import org.apache.spark.{SparkConf, SparkContext}/*** cache()和persist()注意问题* 1.cache()和persist()持久化单位是partition，cache()和persist()是懒执行算子，需要action算子触发执行* 2.对一个RDD使用cache或者persist之后可以赋值给一个变量，下次直接使用这个变量就是使用持久化的数据。* 也可以直接对RDD进行cache或者persist，不赋值给一个变量* 3.如果采用第二种方法赋值给变量的话，后面不能紧跟action算子* 4.cache()和persist()的数据在当前application执行完成之后会自动清除*/
object CacheAndPersist {def main(args: Array[String]): Unit = {val conf: SparkConf = new SparkConf().setMaster("local").setAppName("cache")val sc: SparkContext = new SparkContext(conf)sc.setLogLevel("Error")val lines: RDD[String] = sc.textFile("./data/persistData.txt")// val linescache: RDD[String] = lines.persist(StorageLevel.MEMORY_ONLY)val linescache: RDD[String] = lines.cache()val startTime1: Long = System.currentTimeMillis()val count1: Long = linescache.count()val endTime1: Long = System.currentTimeMillis()println("count1 = "+count1+". time = "+(endTime1-startTime1) + "mm")val starttime2: Long = System.currentTimeMillis()val count2: Long = linescache.count()val endTime2: Long = System.currentTimeMillis()println("count2 = "+count2+", time = "+(endTime2-starttime2) + "ms")sc.stop()}
}14.mapValues算子
针对K，V格式的数据，只对Value做操作，Key保持不变
val infos: RDD[(String, String)] = sc.makeRDD(List[(String,String)](("zhangsan","18"),("lisi","20"),("wangwu","30")))
val result: RDD[(String, String)] = infos.mapValues(s => {s + " " + "zhangsan18"
})
result.foreach(println)
sc.stop()15.flatMapValues算子
（K,V）->(K,V)，作用在K,V格式的RDD上，对一个Key的一个Value返回多个Value
val infos: RDD[(String, String)] = sc.makeRDD(List[(String,String)](("zhangsan","18"),("lisi","20"),("wangwu","30")))val transInfo: RDD[(String, String)] = infos.mapValues(s => {s + " " + "zhangsan18"})
// transInfo.foreach(println)val result: RDD[(String, String)] = transInfo.flatMapValues(s => {//按空格切分s.split(" ")})result.foreach(println)sc.stop()16.combineByKey算子
首先给RDD中每个分区中的每一个key一个初始值
其次在RDD每个分区内部相同的key聚合一次
再次在RDD不同的分区之间将相同的key结果聚合一次
val rdd1: RDD[(String, Int)] = sc.makeRDD(List[(String, Int)](("zhangsan", 10), ("zhangsan", 20), ("wangwu", 30),("lisi", 40), ("zhangsan", 50), ("lisi", 60),("wangwu", 70), ("wangwu", 80), ("lisi", 90)
),3)
rdd1.mapPartitionsWithIndex((index,iter)=>{val arr: ArrayBuffer[(String, Int)] = ArrayBuffer[(String,Int)]()iter.foreach(tp=>{arr.append(tp)println("rdd1 partition index ="+index+".value ="+tp)})arr.iterator
}).count()
println("++++++++++++++++++++++++++++++++++++")
val result: RDD[(String, String)] = rdd1.combineByKey(v=>{v+"hello"}, (s:String, v)=>{s+"@"+v}, (s1:String, s2:String)=>{s1+"#"+s2})
result.foreach(println)17.reduceByKey算子
首先会根据key去分组，然后在每一组中将value聚合，作用在KV格式的RDD上
首先会根据key去分组，然后在每一组中将value聚合，作用在KV格式的RDD上
val infos: RDD[(String, Int)] = sc.parallelize(List[(String,Int)](("zhangsan",1),("zhangsan",2),("zhangsan",3),("lisi",100),("lisi",200)),5)
val result: RDD[(String, Int)] = infos.reduceByKey((v1, v2)=>{v1+v2})
result.foreach(println)
sc.stop()18.repartition算子
重新分区，可以将RDD的分区增多或者减少，会产生shuffle，coalesc(num,true) = repartition(num)
val rdd1: RDD[String] = sc.parallelize(List[String]("love1", "love2", "love3", "love4","love5", "love6", "love7", "love8","love9", "love10", "love11", "love12"
), 3)
val rdd2: RDD[String] = rdd1.mapPartitionsWithIndex((index, iter) => {val list: ListBuffer[String] = ListBuffer[String]()iter.foreach(one => {list.append(s"rdd1 partition = [$index] ,value = [$one]")})list.iterator
}, true)
val rdd3: RDD[String] = rdd2.repartition(3)
val rdd4: RDD[String] = rdd3.mapPartitionsWithIndex((index, iter) => {val arr: ArrayBuffer[String] = ArrayBuffer[String]()iter.foreach(one => {arr.append(s"rdd3 partition = [$index] ,value = [$one]")})arr.iterator
})
val results: Array[String] = rdd4.collect()
results.foreach(println)
sc.stop()19.cogroup算子
合并两个RDD,生成一个新的RDD。分区数与分区数多个那个RDD保持一致
val rdd1 = sc.parallelize(List[(String,String)](("zhangsan","female"),("zhangsan","female1"),("lisi","male"),("wangwu","female"),("maliu","male")),3)
val rdd2 = sc.parallelize(List[(String,Int)](("zhangsan",18),("lisi",19),("lisi",190),("wangwu",20),("tianqi",21)),4)
val resultRDD: RDD[(String, (Iterable[String], Iterable[Int]))] = rdd1.cogroup(rdd2)resultRDD.foreach(info=>{val key = info._1val value1: List[String] = info._2._1.toListval value2: List[Int] = info._2._2.toListprintln("key ="+key+",value"+value1+", value2 = "+value2)
})
println("resultRDD partition length ="+resultRDD.getNumPartitions)
sc.stop()20.join算子
会产生shuffle,(K,V)格式的RDD和(K,V)格式的RDD按照相同的K，join得到(K,(V,W))格式的数据，分区数按照大的来。
val nameRDD: RDD[(String, String)] = sc.parallelize(List[(String,String)](("zhangsan","female"),("lisi","male"),("wangwu","female")),3)
val scoreRDD: RDD[(String, Int)] = sc.parallelize(List[(String,Int)](("zhangsan",18),("lisi",19),("wangwu",20)),2)
val joinRDD: RDD[(String, (String, Int))] = nameRDD.join(scoreRDD)
println(joinRDD.getNumPartitions)
joinRDD.foreach(println)21.leftOutJoin、rightOutJoin算子、fullOuterJoin算子
leftOuterJoin(K,V)格式的RDD和(K,V)格式的RDD，使用leftOuterJoin结合，以左边的RDD出现的key为主 ，得到(K,(V,Option(W)))
val nameRDD: RDD[(String, String)] = sc.parallelize(List[(String,String)](("zhangsan","female"),("lisi","male"),("wangwu","female"),("maliu","male")))
val scoreRDD: RDD[(String, Int)] = sc.parallelize(List[(String,Int)](("zhangsan",22),("lisi",19),("wangwu",20),("tianqi",21)))
val leftOutJoin: RDD[(String, (String, Option[Int]))] = nameRDD.leftOuterJoin(scoreRDD)
leftOutJoin.foreach(println)
sc.stop()rightOuterJoin(K,V)格式的RDD和（K,W）格式的RDD使用rightOuterJoin结合以右边的RDD出现的key为主，得到(K,(Option(V),W))
val nameRDD: RDD[(String, String)] = sc.parallelize(List[(String,String)](("zhangsan","female"),("lisi","male"),("wangwu","female"),("maliu","male")),3)val scoreRDD: RDD[(String, Int)] = sc.parallelize(List[(String,Int)](("zhangsan",18),("lisi",19),("wangwu",20),("tianqi",21)),4)val rightOuterJoin: RDD[(String, (Option[String], Int))] = nameRDD.rightOuterJoin(scoreRDD)rightOuterJoin.foreach(println)println("rightOuterJoin RDD partiotion length = "+rightOuterJoin.getNumPartitions)sc.stop()fullOuterJoin算子(K,,V)格式的RDD和(K,V)格式的RDD，使用fullOuterJoin结合是以两边的RDD出现的key为主，得到(K(Option(V),Option(W)))
val nameRDD: RDD[(String, String)] = sc.parallelize(List[(String,String)](("zhangsan","female"),("lisi","male"),("wangwu","female"),("maliu","male")),3)
val ageRDD: RDD[(String, Int)] = sc.parallelize(List[(String,Int)](("zhangsan",18),("lisi",16),("wangwu",20),("tianqi",21)),4)
val fullOuterJoin: RDD[(String, (Option[String], Option[Int]))] = nameRDD.fullOuterJoin(ageRDD)
fullOuterJoin.foreach(println)
println("fullOuterJoin RDD partition length = "+fullOuterJoin.getNumPartitions)
sc.stop()22.intersection算子
intersection取两个RDD的交集，两个RDD的类型要一致，结果RDD的分区数要与两个父RDD多的那个一致
val rdd1: RDD[String] = sc.parallelize(List[String]("zhangsan","lisi","wangwu"),5)
val rdd2: RDD[String] = sc.parallelize(List[String]("zhangsan","lisi","maliu"),4)
val result: RDD[String] = rdd1.intersection(rdd2)
result.foreach(println)
println("intersection partition length = "+ result.getNumPartitions)
sc.stop()23.foreach算子
foreach遍历RDD中的每一个元素
val lines: RDD[String] = sc.textFile("./data/words") lines.foreach(println)
24.saveAsTextFile算子
将DataSet中的元素以文本的形式写入本地文件系统或者HDFS中，Spark将会对每个元素调用toString方法，将数据元素转换成文本文件中的一行数据，若将文件保存在本地文件系统，那么只会保存在executor所在机器的本地目录
val infos: RDD[String] = sc.parallelize(List[String]("a","b","c","e","f","g"),4)
infos.saveAsTextFile("./data/infos")
25.saveAsObjectFile算子
将数据集中元素以ObjectFile形式写入本地文件系统或者HDFS中
infos.saveAsObjectFile("./data/infosObject")
26.collect算子
collect回收算子，会将结果回收到Driver端，如果结果比较大,就不要回收，这样的话会造成Driver端的OOM
val lines: RDD[String] = sc.textFile("./data/words")
sc.setLogLevel("Error")
val result: Array[String] = lines.collect()
result.foreach(println)27.collectAsMap算子
将K、V格式的RDD回收到Driver端作为Map使用
val weightInfos: RDD[(String, Double)] = sc.parallelize(List[(String,Double)](new Tuple2("zhangsan",99),new Tuple2("lisi",78.6),new Tuple2("wangwu",122.2323)))
val stringToDouble: collection.Map[String, Double] = weightInfos.collectAsMap()
stringToDouble.foreach(tp=>{println(tp._1+"**************"+tp._2)
})
sc.stop()28.count算子
count统计RDD共有多少行数据
val lines: RDD[String] = sc.textFile("./data/sampleData.txt")
val result: Long = lines.count()
println(result)
sc.stop()
直接给出结果行数
29.countByKey算子、countByValue算子
countByKey统计相同的key出现的个数
val rdd: RDD[(String, Integer)] = sc.makeRDD(List[(String,Integer)](("a",1),("a",100),("a",1000),("b",2),("b",200),("c",3),("c",4),("d",122)))
val result: collection.Map[String, Long] = rdd.countByKey()
result.foreach(println)countByValue统计RDD中相同的Value出现的次数，不要求数据必须为RDD格式
val rdd = sc.makeRDD(List[(String,Integer)](("a",1),("a",1),("a",1000),("b",2),("b",200),("c",3),("c",3)))
val result: collection.Map[(String, Integer), Long] = rdd.countByValue()
result.foreach(println)30、take、takeSample算子
take取出RDD中的前N个元素
val lines: RDD[String] = sc.textFile("./data/words")
val array: Array[String] = lines.take(3)
array.foreach(println)
takeSapmle(withReplacement,num,seed)，随机抽样将数据结果拿回Driver端使用，返回Array，
withReplacement:有无放回抽样，num:抽样的条数，seed:种子
val lines: RDD[String] = sc.textFile("./data/words")
val result: Array[String] = lines.takeSample(false,3,10)
result.foreach(println)
31、reduce算子
val rdd: RDD[Int] = sc.makeRDD(Array[Int](1,2,3,4,5))
val result: Int = rdd.reduce((v1, v2) => { v1 + v2 })
//直接得到结果
println(result) }
32.Aggregate算子----transformation类算子
首先是给定RDD的每一个分区一个初始值，然后RDD中每一个分区中按照相同的key，结合初始值去合并，最后RDD之间相同的key聚合
val rdd1: RDD[(String, Int)] = sc.makeRDD(List[(String, Int)](("zhangsan", 10), ("zhangsan", 20), ("wangwu", 30),("lisi", 40), ("zhangsan", 50), ("lisi", 60),("wangwu", 70), ("wangwu", 80), ("lisi", 90)
), 3)
rdd1.mapPartitionsWithIndex((index,iter)=>{val arr: ArrayBuffer[(String, Int)] = ArrayBuffer[(String,Int)]()iter.foreach(tp=>{arr.append(tp)println("rdd1 partition index ="+index+", value ="+tp)})arr.iterator
}).count()
val result: RDD[(String, String)] = rdd1.aggregateByKey("hello")((s, v)=>{s+"~"+v}, (s1, s2)=>{s1+"#"+s2})
result.foreach(println)mapPartitionsWithIndex注释掉执行结果：
33.zip算子 ---Transformation类算子
将两个RDD合成一个K，V格式的RDD，分区数要相同，每个分区中的元素必须相同
val rdd1: RDD[String] = sc.parallelize(List[String]("a","b","c","d"),2)
val rdd2: RDD[Int] = sc.parallelize(List[Int](1,2,3,4),2)
val result: RDD[(String, Int)] = rdd1.zip(rdd2)
result.foreach(println)33、zipWithIndex算子---Transformation类算子
val rdd1 = sc.parallelize(List[String]("a","b","c"),2)
val rdd2 = sc.parallelize(List[Int](1,2,3),numSlices = 2)
val result: RDD[(String, Long)] = rdd1.zipWithIndex()
val result2: RDD[(Int, Long)] = rdd2.zipWithIndex()
result.foreach(println)
result2.foreach(println)spark算子_Spark常用算子相关推荐
- 2021年大数据Spark（十五）：Spark Core的RDD常用算子
目录 常用算子 基本算子 分区操作函数算子 重分区函数算子 1).增加分区函数 2).减少分区函数 3).调整分区函数 聚合函数算子 Scala集合中的聚合函数 ...
- Spark 常用算子详解（转换算子、行动算子、控制算子）
Spark简介 Spark是专为大规模数据处理而设计的快速通用的计算引擎: Spark拥有Hadoop MapReduce所具有的优点,但是运行速度却比MapReduce有很大的提升,特别是在数据挖掘 ...
- 【Spark】Spark的常用算子
Spark的常用算子 目录内容 Spark的常用算子 一.转换算子(Transformation) 二.行动算子(Action) 三.键值对算子(PairRDDFunctions) 四.文件系统算子( ...
- spark 常用算子
一.概述 算子 英文翻译为:Operator(简称op) 狭义:指从一个函数空间到另一个函数空间(或它自身)的映射. 广义:指从一个空间到另一个空间的映射 通俗理解:指事物(数据或函数)从一个状态到另 ...
- 2021年大数据Flink（三十九）：Table与SQL 总结 Flink-SQL常用算子
目录 总结 Flink-SQL常用算子 SELECT WHERE DISTINCT GROUP BY UNION 和 UNION ALL JOI ...
- 2021年大数据Spark（十六）：Spark Core的RDD算子练习
目录 RDD算子练习 map 算子 filter 算子 flatMap 算子 交集.并集.差集.笛卡尔积 distinct 算子 first.take.top 算子 ...
- Spark _07_补充部分算子【二】
接Spark _06_补充部分算子[一] https://blog.csdn.net/qq_41946557/article/details/102673673 scala API package d ...
- spark中各类key算子的用法汇总(持续更新中)
启动方式: spark-shell --master yarn 依赖导入: import org.apache.spark.{SparkConf, SparkContext} 输出rdd的类型举例: ...
- HALCON常用算子（HALCON13.0）
HALCON常用算子(HALCON13.0) Chapter 9--Develop dev_clear_window ( : : : )功能:清除活动图形窗口的内容 dev_close_window ...
最新文章
- RT73 wifi无线网卡驱动移植过程 和wpa支持
- 【双100%提交】剑指 Offer 09. 用两个栈实现队列
- SQL Server遗失管理权限账号密码怎么办？
- 磁共振立体定向仪行业调研报告 - 市场现状分析与发展前景预测（2021-2027年）
- Python加密保护-对可执行的exe进行保护
- Python代码测试 - unitest\doctest\nose\pytest
- DigitalOcean发布弹性块存储服务
- 前端ES6+基础总结
- 【Grub Grub2】万能优盘启动盘 （WinPE、LinuxPE）-- 方法1 U盘三分区法（不推荐，供参考）
- Misra c规则简介
- 安卓手机开机动画bootanimation.zip文件制作以及注意事项
- 小程序源码图片列表，图片下载
- linux中一次显示一页内容的命令
- 中国医师节丨华为IdeaHub用远程诊疗护佑人民健康，为医生减负
- 如何通过轨迹信息判断驾驶人是否为同一人？
- 运维之道 | Nginx调优
- Dubbo 原理和机制详解
- js如何往数组Array（list）中添加元素
- ARCore快速入门--使用GLSurfaceView和ARCore绘制3D模型
- 新概念英语1册51课