189 8069 5689

Spark操作中的aggregate、aggregateByKey怎么理解

今天就跟大家聊聊有关Spark操作中的之aggregate、aggregateByKey怎么理解,可能很多人都不太了解,为了让大家更加了解,小编给大家总结了以下内容,希望大家根据这篇文章可以有所收获。

在安乡等地区,都构建了全面的区域性战略布局,加强发展的系统性、市场前瞻性、产品创新能力,以专注、极致的服务理念,为客户提供做网站、网站设计 网站设计制作按需设计网站,公司网站建设,企业网站建设,成都品牌网站建设,成都全网营销推广,外贸营销网站建设,安乡网站建设费用合理。

1. aggregate函数

将每个分区里面的元素进行聚合,然后用combine函数将每个分区的结果和初始值(zeroValue)进行combine操作。这个函数最终返回的类型不需要和RDD中元素类型一致。

seqOp操作会聚合各分区中的元素,然后combOp操作把所有分区的聚合结果再次聚合,两个操作的初始值都是zeroValue. seqOp的操作是遍历分区中的所有元素(T),第一个T跟zeroValue做操作,结果再作为与第二个T做操作的zeroValue,直到遍历完整个分区。combOp操作是把各分区聚合的结果,再聚合。aggregate函数返回一个跟RDD不同类型的值。因此,需要一个操作seqOp来把分区中的元素T合并成一个U,另外一个操作combOp把所有U聚合。

例子程序:

scala> val rdd = List(1,2,3,4,5,6,7,8,9)rdd: List[Int] = List(1, 2, 3, 4, 5, 6, 7, 8, 9)scala> rdd.par.aggregate((0,0))((acc,number) => (acc._1 + number, acc._2 + 1),(par1,par2) => (par1._1 + par2._1, par1._2 + par2._2))res0: (Int, Int) = (45,9)scala> res0._1 / res0._2res1: Int = 5

过程大概这样:

首先,初始值是(0,0),这个值在后面2步会用到。然后,(acc,number) => (acc._1 + number, acc._2 + 1),number即是函数定义中的T,这里即是List中的元素。所以acc._1 + number,acc._2 + 1的过程如下。

1. 0+1, 0+12. 1+2, 1+13. 3+3, 2+14. 6+4, 3+15. 10+5, 4+16. 15+6, 5+17. 21+7, 6+18. 28+8, 7+19. 36+9, 8+1

结果即是(45,9)。这里演示的是单线程计算过程,实际Spark执行中是分布式计算,可能会把List分成多个分区,假如3个,p1(1,2,3,4),p2(5,6,7,8),p3(9),经过计算各分区的的结果(10,4),(26,4),(9,1),这样,执行(par1,par2) =>(par1._1 + par2._1, par1._2 + par2._2)就是(10+26+9,4+4+1)即(45,9),再求平均值就简单了。

2. aggregateByKey函数:

对PairRDD中相同的Key值进行聚合操作,在聚合过程中同样使用了一个中立的初始值。和aggregate函数类似,aggregateByKey返回值的类型不需要和RDD中value的类型一致。因为aggregateByKey是对相同Key中的值进行聚合操作,所以aggregateByKey'函数最终返回的类型还是PairRDD,对应的结果是Key和聚合后的值,而aggregate函数直接返回的是非RDD的结果。

例子程序:

import org.apache.spark.SparkConfimport org.apache.spark.SparkContextobject AggregateByKeyOp { def main(args:Array[String]){   val sparkConf: SparkConf = new SparkConf().setAppName("AggregateByKey").setMaster("local")  val sc: SparkContext = new SparkContext(sparkConf)      val data=List((1,3),(1,2),(1,4),(2,3))   val rdd=sc.parallelize(data, 2)      

//合并不同partition中的值,a,b得数据类型为zeroValue的数据类型   def combOp(a:String,b:String):String={    println("combOp: "+a+"\t"+b)    a+b   }   

//合并在同一个partition中的值,a的数据类型为zeroValue的数据类型,b的数据类型为原value的数据类型   def seqOp(a:String,b:Int):String={    println("SeqOp:"+a+"\t"+b)    a+b   }   rdd.foreach(println)   

//zeroValue:中立值,定义返回value的类型,并参与运算   

//seqOp:用来在同一个partition中合并值   

//combOp:用来在不同partiton中合并值   

val aggregateByKeyRDD=rdd.aggregateByKey("100")(seqOp, combOp)   sc.stop() }}

运行结果:

将数据拆分成两个分区

//分区一数据(1,3)(1,2)//分区二数据(1,4)(2,3)

//分区一相同key的数据进行合并seq: 100 3 

//(1,3)开始和中立值进行合并 合并结果为 1003seq: 1003 2 //(1,2)再次合并 结果为 10032

//分区二相同key的数据进行合并seq: 100 4 

//(1,4) 开始和中立值进行合并 1004seq: 100 3 //(2,3) 开始和中立值进行合并 1003

将两个分区的结果进行合并/

/key为2的,只在一个分区存在,不需要合并 (2,1003)(2,1003)

//key为1的, 在两个分区存在,并且数据类型一致,合并comb: 10032 1004(1,100321004)

看完上述内容,你们对Spark操作中的之aggregate、aggregateByKey怎么理解有进一步的了解吗?如果还想了解更多知识或者相关内容,请关注创新互联行业资讯频道,感谢大家的支持。


分享题目:Spark操作中的aggregate、aggregateByKey怎么理解
本文地址:http://cdxtjz.cn/article/ghjchh.html

其他资讯