10.3 spark算子例子map,filter,flatMap,mapPartitions,groupByKey,join,distinct

快来打我* 2022-05-19 03:56 182阅读 0赞

Map filter:

object Map\_Operator \{

def main(args: Array\[String\]): Unit = \{

/\*\*

\* 创建一个设置Spark运行参数的对象

\* SparkConf对象可以设置运行模式，设置Application的名称

\* 设置Application执行所需要的资源情况

\*/

val conf = new SparkConf().setMaster("local").setAppName("Map\_Operator")

/\*\*

\* 创建一个SparkContext的上下文对象

\* SparkContext是通往集群的 唯一通道

\* 负责任务分发，以及任务失败后的重试工作

\*/

val sc = new SparkContext(conf)

/\*\*

\* makeRDD方法的第一个参数代表的是RDD中的 元素

\* 第二个参数：RDD的分区数

\* rdd\[Int\]

\*/

val rdd = sc.makeRDD(1 to 10,3)

/\*\*

\* mapRDD\[String\]

\* Map算子是一条一条的计算的

\*/

val mapRDD = rdd.map \{\_+"~"\}

/\*\*

\* 想要执行，必须有action类的算子

\* collect算子会将集群中计算的结果回收到Driver端，慎用

\*/

val resultArr = mapRDD.collect()

resultArr.foreach \{ println \}

rdd.filter \{ x => x!=2 \}.foreach \{ println \}

/\*\*

\* 释放资源

\*/

sc.stop()

--------------------

flatMap:

val rdd = sc.parallelize(Array("hello,bjsxt","hello,xuruyun"),1)

rdd.map \{ x => x.split(",") \}.foreach \{ println \}

val flatMapRDD = rdd.flatMap \{ x => x.split(",") \}

flatMapRDD.foreach \{ println \}

sc.stop()

--------------------

mapPartitions:

/\*\*

\* makeRDD方法的第一个参数代表的是RDD中的 元素

\* 第二个参数：RDD的分区数

\* rdd\[Int\]

\*/

val rdd = sc.makeRDD(1 to 10,3)

/\*\*

\* mapPartitions这个算子遍历的单位是partition

\* 会将一个partition的数据量全部加载到一个集合里面

\*/

val mapPartitonsRDD = rdd.mapPartitions(iterator=>\{

val list = new ListBuffer\[Int\]()

//创建一个数据库连接

while(iterator.hasNext)\{

val num = iterator.next()

list.+=(num+100)

//批量插入数据库

list.iterator

\}, false)

/\*\*

\* 想要执行，必须有action类的算子

\* collect算子会将集群中计算的结果回收到Driver端，慎用

\*/

val resultArr = mapPartitonsRDD.collect()

resultArr.foreach \{ println \}

--------------------

mapPartitionsWithIndex:

/\*\*

\* makeRDD方法的第一个参数代表的是RDD中的 元素

\* 第二个参数：RDD的分区数

\* rdd\[Int\]

\*/

val rdd = sc.makeRDD(1 to 10,3)

/\*\*

\* mapPartitions这个算子遍历的单位是partition

\* 会将一个partition的数据量全部加载到一个集合里面

\*/

val mapPartitionsWithIndexRDD = rdd.mapPartitionsWithIndex((index,iterator)=>\{

val list = new ListBuffer\[Int\]()

while (iterator.hasNext) \{

val num = iterator.next()

println("partitionId:" + index + "value:" + num) //打印分区

list+=num

list.iterator

\}, false).collect()

--------------------

GroupByKey：

val rdd1 = sc.makeRDD(Array(

Tuple2("A",1),

("B",2),

("B",5),

("C",3),

Tuple2("D",4)

))

/\*\*

\* A \[1\]

\* B \[2,5\]

\* C\[3\]

\*/

val grdd = rdd1.groupByKey()

rdd1.reduceByKey((a,b)=>a+b).collect().foreach(println)

--------------------

Join：

val rdd1 = sc.makeRDD(Array(

Tuple2("A",1),

("B",2),

("B",5),

("C",3),

Tuple2("D",4)

))

val rdd2 = sc.makeRDD(Array(

Tuple2("A",11),

Tuple2("B",22),

Tuple2("C",33),

Tuple2("D",44),

Tuple2("E",55)

))

/\*\*

\* join就是一个内连接

\*/

val joinResultRDD = rdd2.join(rdd1)

joinResultRDD.take(5).foreach(println)

val result = joinResultRDD.first()

println("key:" + result.\_1 + "RDD2 value:" + result.\_2.\_1 + "RDD2 value:"+ result.\_2.\_2)

rdd2.leftOuterJoin(rdd1).foreach(println)

--------------------

Distinct：

val rdd1 = sc.makeRDD(Array(

("A",1),

("B",2),

("B",5),

("C",3),

("D",4)

))

val resultArr = rdd1.map(x=>(x,null)).groupByKey().collect() //这句就相当于distinct

resultArr.foreach(x=>println(x.\_1))

10.3 spark算子例子map,filter,flatMap,mapPartitions,groupByKey,join,distinct

发表评论取消回复

还没有评论，来说两句吧...

相关阅读

相关 Spark 算子

相关算子实现_spark算子调优

相关 Spark算子实现WordCount

相关 spark 算子

相关 Spark算子总结

相关图解Spark Transformation算子

相关 Spark 基本算子

相关 Spark、RDD、算子

相关 Spark算子代码汇总

相关 spark算子集合