dxyadc 2019-10-28
目录
@
基于上次的wordcount,我们来写一个wordcountcount,来对wc程序进行第二次计数,我们来分析一下性能。
package com.littlepage.wc import org.apache.spark.rdd.RDD import org.apache.spark.{SparkConf, SparkContext} object WordCount { def main(args: Array[String]): Unit = { val conf=new SparkConf().setAppName("wc").setMaster("local") val sparkContext=new SparkContext(conf) sparkContext.setLogLevel("error") val fileRDD:RDD[String] = sparkContext.textFile("data/data") val words:RDD[String] = fileRDD.flatMap(_.split(" ")) val pairWord:RDD[(String,Int)] = words.map((_,1)) val res:RDD[(String,Int)] = pairWord.reduceByKey(_+_) println("wordcount:") res.foreach(println) val rev:RDD[(Int,Int)] = res.map((x)=>{(x._2,1)}) val pl:RDD[(Int,Int)] = rev.reduceByKey(_+_) println("\nwordcountcount") pl.foreach(println) Thread.sleep(100000000) } }
通过性能图,我们可以知道:
1.Spark如果不对其结果进行存储或输出,那么Spark将不会处理map或者reduce操作
2.如果进行重复输出,共用的map或者reduce操作只执行一次
3.默认如果产生一次shuffle是去查看图表的一次拐弯,为了尽量减少性能的消耗,编写程序时应该尽量减少shuffle的次数
Spark编程模型和MapReduce相比,Spark可以多个Job,多个State进行执行。
源码部分参考视频
我们在写一个Spark程序中,不可避免的算子有三个,创建算子,转换算子,收集算子。
创建算子可以创建一个RDD数据集,这个创建可以在内存中(集合容器),也可以在硬盘中(文件)获取
转换算子可以处理一个RDD数据集,即map和reduce操作,都算做转换算子。
收集算子我们在写一个RDD数据集的时候,必须使用收集算子进行收集,否则不会触发shuffle。
示例,三个算子写一个过滤数字程序。
package com.littlepage import org.apache.spark.rdd.RDD import org.apache.spark.{SparkConf, SparkContext} object demo2 { def main(args: Array[String]): Unit = { val conf=new SparkConf().setAppName("demo2").setMaster("local") val sc=new SparkContext(conf) sc.setLogLevel("error") val dataRDD: RDD[Int] = sc.parallelize(List(1,2,3,4,5,6,7,6,5,4,3,2,1))//创建算子 val filterRDD: RDD[Int] = dataRDD.filter(_>3)//转换算子 val ints:Array[Int] = filterRDD.collect()//收集算子 Thread.sleep(100000) } }
package com.littlepage import org.apache.spark.rdd.RDD import org.apache.spark.{SparkConf, SparkContext} object demo2 { def main(args: Array[String]): Unit = { val conf=new SparkConf().setAppName("demo2").setMaster("local") val sc=new SparkContext(conf) sc.setLogLevel("error") val dataRDD: RDD[Int] = sc.parallelize(List(1,2,3,4,5,6,7,6,5,4,3,2,1))//创建算子 val filterRDD: RDD[Int] = dataRDD.filter(_>3)//转换算子 val ints:Array[Int] = filterRDD.collect()//收集算子 Thread.sleep(100000) } }
2.1.union算子
将两个数据集合并为一个数据集,直接合并,不会产生shuffle
object union { def main(args: Array[String]): Unit = { val sc=new SparkContext(new SparkConf().setMaster("local").setAppName("union")) sc.setLogLevel("error") val rdd1:RDD[Int] = sc.parallelize(List(1,2,3,4,6,7)) val rdd2:RDD[Int] = sc.parallelize(List(2,3,4,5)) val uniondata = rdd1.union(rdd2) uniondata.foreach(print) Thread.sleep(100000) } }
2.2.intersection算子
将2个数据集取交集,产生一个shuffle
val interdata:RDD[Int] = rdd1.intersection(rdd2)
2.3.substract算子
将2个数据集取差集,产生一个shuffle
val subdata:RDD[Int] = rdd1.substract(rdd2)
2.4.cartesian算子
将2个数据集取笛卡尔积,不产生shuffle
val cartesiandata:RDD[Int] = rdd1.cartesian(rdd2)
2.5.cogroup算子
两个分组进行,key作为结果的key,value集合进行一个二元祖,包含两个分区的元素,产生一个shuffle。
val rdd1:RDD[(String,Int)] = sc.parallelize(List( ("zhangsan",11), ("zhangsan",12), ("lisi",13), ("wangwu",14) )); val rdd2:RDD[(String,Int)] = sc.parallelize(List( ("zhangsan",21), ("zhangsan",22), ("lisi",23), ("zhaoliu",28) )) val cogroupdata:RDD[(String,(Iterable[Int],Iterable[Int]))] = rdd1.cogroup(rdd2)
6.join,leftOuterJoin,rightOuterJoin,fullOuterJoin算子
val joindata:RDD[(String,(Int,Int))] = rdd1.join(rdd2) val leftdata:RDD[(String,(Int,Option[Int]))] = rdd1.leftOuterJoin(rdd2) val rightdata:RDD[(String,(Option[Int],Int))] = rdd2.rightOuterJoin(rdd2) val fulldata:RDD[(String,(Option[Int],Option[Int]))] = rdd1.fullOuterJoin(rdd2)
3.1.swap算子
将一个k-v数据集的key和value交换,用法
data.map(_.swap)
3.2.sort算子
sort算子可以将按照key进行全排序
data.sortByKey()
3.3.take算子
获得数据的前n个,n为一个整型
data.take(n)
3.4.distinct去重
去除key相同的
val keys:RDD[(String,String) = map.distinct()