Spark为什么只有在调用action时才会触发任务执行呢(附算子优化和使用示例)?

编程之家收集整理的这篇文章主要介绍了Spark为什么只有在调用action时才会触发任务执行呢(附算子优化和使用示例)?编程之家小编觉得挺不错的,现在分享给大家,也给大家做个参考。

搜索热词

Spark算子主要划分为两类:transformation和action,并且只有action算子触发的时候才会真正执行任务。还记得之前的文章《Spark RDD详解》中提到,Spark RDD的缓存和checkpoint是懒加载操作,只有action触发的时候才会真正执行,其实不仅是Spark RDD,在Spark其他组件如SparkStreaming中也是如此,这是Spark的一个特性之一。像我们常用的算子map、flatMap、filter都是transformation算子,而collect、count、saveAsTextFile、countByKey、foreach则为action算子。

 

但初学Spark的人往往都会有这样的疑惑,为什么Spark任务只有在调用action算子的时候,才会真正执行呢?咱们来假设一种情况:假如Spark中transformation直接触发Spark任务!那么会产生什么结果呢?

 

1. 导致map执行完了要立即输出,数据也必然要落地(内存和磁盘)

 

2. map任务的生成、调度、执行,以及彼此之间的rpc通信等等,当牵扯到大量任务、大数据量时,会很影响性能

看到这两点是不是很容易联想到MapReduce的计算模型,MapReduce因为中间结果需要落地,导致性能相对Spark较低下,这也是MapReduce广为诟病的原因之一。所以Spark采用只有调用action算子时才会真正执行任务,这是相对于MapReduce的优化点之一。

但是每个Spark RDD中连续调用多个map类算子,Spark任务是对数据在一次循环遍历中完成还是每个map算子都进行一次循环遍历呢?

答案很确定:不需要对每个map算子都进行循环遍历。Spark会将多个map算子pipeline起来应用到RDD分区的每个数据元素上(后续将要介绍的Sparksql中的Dataset/DataFrame也是如此)

下面说几个算子的优化,这也是面试中经常问的问题:

在我们实际的业务场景中经常会使用到根据key进行分组聚合的操作,当然熟悉Spark算子使用的都知道像reduceByKey、groupByKey、aggregateByKey、combineByKey大多都能满足需求。但是笔者在这里还是要重点说一下,因为很多人想到分组聚合往往第一个想到的算子就是groupByKey,但是groupByKey相对其他算子性能低并且处理不好的情况下,容易发生数据倾斜。所以我们能用其他算子比如reduceByKey替代groupByKey实现满足我们业务需求的,就一律不用groupByKey。当然reduceByKey在某些场景下性能会比aggregateByKey低,具体算子的替换要结合实际业务需求场景来定。

这里主要说明一下reduceByKey和groupByKey的对比,以及几个算子替代的场景示例:

1.首先这几个“ByKey”的算子会触发shullfe,这里强调一点,对于分布式任务,如果存在聚合操作的话往往都是要进行shuffle的

2.相对于reduceByKey,groupByKey没有预先聚合,而是直接将相同key的value进行分组然后再聚合造成shuffle耗费严重;而reduceByKey会先在map端进行局部聚合,然后再在reduce端再次聚合,这点类似于MapReduce中combiner组件,可以减少磁盘IO和网络IO,提高性能

3.aggregateByKey替代reduceByKey的场景:当输出的结果和输入的结果不同的时候可以被替换。例如,查找同一个key的所有不同的value值,也即是先根据key进行分组,然后去重。假设采用reduceByKey实现的话,需要先用map讲单个元素装到set里,然后在针对set进行reduceByKey,伪代码:rdd.map(case(k,v) => (k,Set(v))).reduceByKey(_ ++ _),但是该过程会导致为每个记录创建一个set,这是很没必要的。此时我们可以使用aggregateByKey替代reduceByKey实现该需求,伪代码

val zero = mutable.Set[String]() 
rdd.aggregateByKey(zero)((set,v) => set += v,(set1,set2) => set1 ++= set2)。

具体示例:

1)reduceByKey

val rdd = rowRdd.map { row =>

      val id = row.getAs[String]("id")

      val name = row.getAs[String]("name")

      val count = row.getAs[Long]("count")

      (id,(name,count))

    }.map { case (id,count)) => (id,Array(count)) }.reduceByKey(_ ++ _)

 

2)aggregateByKey

 val zeroValue = mutable.Set[(String,Long)]()

 val rdd = df.rdd.map { row =>

      val id = row.getAs[String]("id")

      val name = row.getAs[String]("name")

      val count = row.getAs[Long]("count")

      (id,count))

 }.aggregateByKey(zeroValue)(

      (set,

      (set1,set2) => set1 ++= set2)

 

3combineByKey

val rdd = df.rdd.map { row =>

      val id = row.getAs[String]("id")

      val name = row.getAs[String]("name")

      val count = row.getAs[Long]("count")

      (id,count))

}.combineByKey(

      (v: (String,Long)) => List(v),

      (c: List[(String,Long)],v: (String,Long)) => v :: c,

      (c1: List[(String,c2: List[(String,Long)]) => c1 ::: c2)

 

 

4.当两个数据集已经按照key进行分组,此时想对两个数据集在仍然保持分组的基础上进行join,则可以使用cgroup,以避免分组展开然后再次分组的开销

Spark目前提供了80多种算子,想熟练掌握这些算子如何运用,笔者建议学习一下Scala语言,原因除了《Spark通识中说的那两点之外,还有就是Spark提供的很多算子跟Scala本身提供的函数功能很相似甚至有些名字都是一样的,了解了Scala提供的,对于学习Spark算子将事半功倍。这里举一些常用的transformation和action使用示例:

 

transformation

 

>> map

map是对RDD中的每个元素都执行一个指定的函数来产生一个新的RDD。任何原RDD中的元素在新RDD中都有且只有一个元素与之对应。

举例:

val a = sc.parallelize(1 to 9,3)

val b = a.map(x => x*2)

a.collect 【Array[Int] = Array(1,2,3,4,5,6,7,8,9)】

b.collect 【Array[Int] = Array(2,10,12,14,16,18)】

 

>> filter

filter是对RDD中的每个元素都执行一个指定的函数来过滤产生一个新的RDD,该RDD由经过函数处理后返回值为true的输入元素组成。任何原RDD中的元素在新RDD中都有且只有一个元素与之对应。

val rdd = sc.parallelize(List(1,6))

val filterRdd = rdd.filter(_ > 3)

filterRdd.collect() 【返回所有大于3的数据的:Array(6,12)】

 

>> flatMap

与map类似,区别是原RDD中的元素经map处理后只能生成一个元素,而原RDD中的元素经flatmap处理后可生成多个元素来构建新RDD。举例:对原RDD中的每个元素x产生y个元素(从1到y,y为元素x的值)

val a = sc.parallelize(1 to 4,2)

val b = a.flatMap(x => 1 to x)

b.collect 【Array(1,1,4)】

 

>> reduceByKey和sortByKey

        分组聚合与排序,这里以单词统计,并按单词排序为例

val rdd1 = sc.parallelize(List(("tom",1),("jerry",3),("kitty",2),("shuke",1)))

val rdd2 = sc.parallelize(List(("jerry",("tom",5)))

val rdd3 = rdd1.union(rdd2)

//按key进行聚合(PairRDDFunctions提供)

val rdd4 = rdd3.reduceByKey(_ + _)

//false降序,默认true(OrderedRDDFunctions提供)

val rdd5 = rdd4.sortByKey(false)

 

>> repartition

函数其实就是coalesce函数第二个参数为true的实现,改变分区数会产生shuffle,repartition之后会返回一个新的RDD

var data = sc.parallelize(1 to 12,3) //分区数3

var rdd1 = data.repartition(1) //分区数1

var rdd1 = data.repartition(4) //4

data.partitions.size 还是3

 

action

 

>> first

first返回RDD中的第一个元素,不排序。

var rdd1 = sc.makeRDD(Array(("A","1"),("B","2"),("C","3")),2)

rdd1.first 【 (A,1) 】

var rdd1 = sc.makeRDD(Seq(10,3))

rdd1.first 【 10 】

 

>> count

        count返回RDD中的元素数量

var rdd1 = sc.makeRDD(Array(("A",2)

rdd1.count 【 3 】

 

>> take

take用于获取RDD中从0到num-1下标的元素,不排序。

var rdd1 = sc.makeRDD(Seq(10,3))
rdd1.take(1) 【 Array(10) 】
rdd1.take(2) 【 Array(10,4) 】

 

像各种save操作,如saveAsNewAPIHadoopDataset都是action算子,这里就不一一列举了。


关注微信公众号:大数据学习与分享获取更对技术干货

总结

以上是编程之家为你收集整理的Spark为什么只有在调用action时才会触发任务执行呢(附算子优化和使用示例)?全部内容,希望文章能够帮你解决Spark为什么只有在调用action时才会触发任务执行呢(附算子优化和使用示例)?所遇到的程序开发问题。

如果觉得编程之家网站内容还不错,欢迎将编程之家网站推荐给程序员好友。

本图文内容来源于网友网络收集整理提供,作为学习参考使用,版权属于原作者。
如您喜欢寻找一群志同道合、互帮互助的学习伙伴,可以点击下方链接加入:
编程之家官方1群:1065694478(已满)
编程之家官方2群:163560250(已满)
编程之家官方3群:312128206(已满)
编程之家官方4群:230427597

相关文章

猜你在找的Spark相关文章

SparkStreaming实时消费kafka数据,结合redis实时统计pv,uv,结果保存到mysql。
RDD(Resilient Distributed Datasets)弹性的分布式数据集,又称Spark core,它代表一个只读的、不可变、可分区,里面的元素可分布式并行计算的数据集。RDD是一个很
最近经常有小伙伴留言,核心问题都比较类似,就是虽然接触Spark有一段时间了,但是搞不明白一个问题,为什么我从HDFS上加载不同的文件时,打印的分区数不一样,并且好像spark.default.par
Spark SQL是一个用来处理结构化数据的Spark组件,前身是shark,但是shark过多的依赖于hive如采用hive的语法解析器、查询优化器等,制约了Spark各个组件之间的相互集成,因此S
Spark闭包 | driver & executor程序代码执行 ==> 编写的Spark程序代码,运行在driver端还是executor端呢?
Spark算子主要划分为两类:transformation和action,并且只有action算子触发的时候才会真正执行任务。还记得之前的文章《Spark RDD详解》中提到,Spark RDD的缓存
对于Spark开发人员来说,一个比较普遍的问题就是如何合理的配置Spark的硬件?当然如何合理的对Spark集群进行硬件配置要视情况而定,在这里给出一些建议
Apache Spark是一种快速、通用、可扩展、可容错的、基于内存迭代计算的大数据分析引擎。首先强调一点, Spark目前是一个处理数据的计算引擎, 不做存储。首先咱们通过一张图来看看目前Spark