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._2
res1: 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+1
2.  1+2,  1+1
3.  3+3,  2+1
4.  6+4,  3+1
5.  10+5,  4+1
6.  15+6,  5+1
7.  21+7,  6+1
8.  28+8,  7+1
9.  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.SparkConf
import org.apache.spark.SparkContext

object 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)開始和中立值進行合并  合并結果為 1003
seq: 1003     2   //(1,2)再次合并 結果為 10032

//分區二相同key的數據進行合并
seq: 100     4  //(1,4) 開始和中立值進行合并 1004
seq: 100     3  //(2,3) 開始和中立值進行合并 1003

將兩個分區的結果進行合并
//key為2的，只在一個分區存在，不需要合并 (2,1003)
(2,1003)

//key為1的, 在兩個分區存在，并且數據類型一致，合并
comb: 10032     1004
(1,100321004)

以上就是本文的全部內容，希望對大家的學習有所幫助，也希望大家多多支持億速云。