Spark經常使用算子詳解

Spark的算子的分類

　　　從大方向來講，Spark 算子大體能夠分爲如下兩類:

     1）Transformation 變換/轉換算子：這種變換並不觸發提交做業，完成做業中間過程處理。

　　　　　Transformation 操做是延遲計算的，也就是說從一個RDD 轉換生成另外一個 RDD 的轉換操做不是立刻執行，須要等到有 Action 操做的時候纔會真正觸發運算。

     2）Action 行動算子：這類算子會觸發 SparkContext 提交 Job 做業。

　　  　　Action 算子會觸發 Spark 提交做業（Job），並將數據輸出 Spark系統。

 

　　從小方向來講，Spark 算子大體能夠分爲如下三類:

　　1）Value數據類型的Transformation算子，這種變換並不觸發提交做業，針對處理的數據項是Value型的數據。
　　2）Key-Value數據類型的Transfromation算子，這種變換並不觸發提交做業，針對處理的數據項是Key-Value型的數據對。

　　3）Action算子，這類算子會觸發SparkContext提交Job做業。

 

 

1）Value數據類型的Transformation算子　　

　　1、輸入分區與輸出分區一對一型

　　　　一、map算子

　　　　二、flatMap算子

　　　　三、mapPartitions算子

　　　　四、glom算子

　　2、輸入分區與輸出分區多對一型　

　　　　五、union算子

　　　　六、cartesian算子

　　3、輸入分區與輸出分區多對多型

　　　　七、grouBy算子

　　4、輸出分區爲輸入分區子集型

　　　　八、filter算子

　　　　九、distinct算子

　　　　十、subtract算子

　　　　十一、sample算子

     　　 十二、takeSample算子

 　　5、Cache型

　　　　1三、cache算子　　

　　　　1四、persist算子

 

2）Key-Value數據類型的Transfromation算子

　　1、輸入分區與輸出分區一對一

　　　　1五、mapValues算子

　　2、對單個RDD或兩個RDD彙集

　　　單個RDD彙集

　　　　1六、combineByKey算子

　　　　1七、reduceByKey算子

　　　　1八、partitionBy算子

 　　兩個RDD彙集

　　　　1九、Cogroup算子

　　3、鏈接

　　　　20、join算子

　　　　2一、leftOutJoin和 rightOutJoin算子

 

 

 3）Action算子

　　1、無輸出

　　　　2二、foreach算子

　　2、HDFS

　　　　2三、saveAsTextFile算子

　　　　2四、saveAsObjectFile算子

　　3、Scala集合和數據類型

　　　　2五、collect算子

　　　　2六、collectAsMap算子

 　　　  2七、reduceByKeyLocally算子

 　　　  2八、lookup算子

　　　　2九、count算子

　　　　30、top算子

　　　　3一、reduce算子

　　　　3二、fold算子

　　　　3三、aggregate算子

 

 

     1. Transformations 算子

　（1） map

　　將原來 RDD 的每一個數據項經過 map 中的用戶自定義函數 f 映射轉變爲一個新的元素。源碼中 map 算子至關於初始化一個 RDD， 新 RDD 叫作 MappedRDD(this, sc.clean(f))。

     圖 1中每一個方框表示一個 RDD 分區，左側的分區通過用戶自定義函數 f:T->U 映射爲右側的新 RDD 分區。可是，實際只有等到 Action算子觸發後，這個 f 函數纔會和其餘函數在一個stage 中對數據進行運算。在圖 1 中的第一個分區，數據記錄 V1 輸入 f，經過 f 轉換輸出爲轉換後的分區中的數據記錄 V’1。
                            

 

　　　　　　圖1    map 算子對 RDD 轉換　　　　　　　　　　　　　　　　　　　

 

    （2） flatMap

     將原來 RDD 中的每一個元素經過函數 f 轉換爲新的元素，並將生成的 RDD 的每一個集合中的元素合併爲一個集合，內部建立 FlatMappedRDD(this，sc.clean(f))。
　　圖 2 表 示 RDD 的 一 個 分 區 ，進 行 flatMap函 數 操 做， flatMap 中 傳 入 的 函 數 爲 f:T->U， T和 U 能夠是任意的數據類型。將分區中的數據經過用戶自定義函數 f 轉換爲新的數據。外部大方框能夠認爲是一個 RDD 分區，小方框表明一個集合。 V一、 V二、 V3 在一個集合做爲 RDD 的一個數據項，可能存儲爲數組或其餘容器，轉換爲V’一、 V’二、 V’3 後，將原來的數組或容器結合拆散，拆散的數據造成爲 RDD 中的數據項。

 

 

　　　　　　　　圖2    　flapMap 算子對 RDD 轉換

    （3） mapPartitions

      mapPartitions 函 數 獲 取 到 每 個 分 區 的 迭 代器，在 函 數 中 通 過 這 個 分 區 整 體 的 迭 代 器 對整 個 分 區 的 元 素 進 行 操 做。 內 部 實 現 是 生 成
MapPartitionsRDD。圖 3 中的方框表明一個 RDD 分區。圖 3 中，用戶經過函數 f (iter)=>iter.f ilter(_>=3) 對分區中全部數據進行過濾，大於和等於 3 的數據保留。一個方塊表明一個 RDD 分區，含有 一、 二、 3 的分區過濾只剩下元素 3。

 

 

　　　　圖3  mapPartitions 算子對 RDD 轉換

 

　　（4）glom

　　glom函數將每一個分區造成一個數組，內部實現是返回的GlommedRDD。 圖4中的每一個方框表明一個RDD分區。圖4中的方框表明一個分區。 該圖表示含有V一、 V二、 V3的分區經過函數glom造成一數組Array[（V1），（V2），（V3）]。

 

 

　　　　　　圖 4   glom算子對RDD轉換

 

     （5） union

      使用 union 函數時須要保證兩個 RDD 元素的數據類型相同，返回的 RDD 數據類型和被合併的 RDD 元素數據類型相同，並不進行去重操做，保存全部元素。若是想去重
可使用 distinct()。同時 Spark 還提供更爲簡潔的使用 union 的 API，經過 ++ 符號至關於 union 函數操做。
     圖 5 中左側大方框表明兩個 RDD，大方框內的小方框表明 RDD 的分區。右側大方框表明合併後的 RDD，大方框內的小方框表明分區。

　　含有V一、V二、U一、U二、U三、U4的RDD和含有V一、V八、U五、U六、U七、U8的RDD合併全部元素造成一個RDD。V一、V一、V二、V8造成一個分區，U一、U二、U三、U四、U五、U六、U七、U8造成一個分區。

 

　

　圖 5  union 算子對 RDD 轉換　

 

　　（6） cartesian

       對 兩 個 RDD 內 的 所 有 元 素 進 行 笛 卡 爾 積 操 做。 操 做 後， 內 部 實 現 返 回CartesianRDD。圖6中左側大方框表明兩個 RDD，大方框內的小方框表明 RDD 的分區。右側大方框表明合併後的 RDD，大方框內的小方框表明分區。圖6中的大方框表明RDD，大方框中的小方框表明RDD分區。
      例 如： V1 和 另 一 個 RDD 中 的 W一、 W二、 Q5 進 行 笛 卡 爾 積 運 算 形 成 (V1,W1)、(V1,W2)、 (V1,Q5)。
     

 

       圖 6  cartesian 算子對 RDD 轉換

 

　　（7） groupBy

　　groupBy ：將元素經過函數生成相應的 Key，數據就轉化爲 Key-Value 格式，以後將 Key 相同的元素分爲一組。
　　函數實現以下：
　　1）將用戶函數預處理：
　　val cleanF = sc.clean(f)
　　2）對數據 map 進行函數操做，最後再進行 groupByKey 分組操做。

     this.map(t => (cleanF(t), t)).groupByKey(p)
　　其中， p 肯定了分區個數和分區函數，也就決定了並行化的程度。

　　圖7 中方框表明一個 RDD 分區，相同key 的元素合併到一個組。例如 V1 和 V2 合併爲 V， Value 爲 V1,V2。造成 V,Seq(V1,V2)。

 

　　圖 7 groupBy 算子對 RDD 轉換

 

　　（8） filter

    filter 函數功能是對元素進行過濾，對每一個 元 素 應 用 f 函 數， 返 回 值 爲 true 的 元 素 在RDD 中保留，返回值爲 false 的元素將被過濾掉。 內 部 實 現 相 當 於 生 成 FilteredRDD(this，sc.clean(f))。
    下面代碼爲函數的本質實現：
    deffilter(f:T=>Boolean):RDD[T]=newFilteredRDD(this,sc.clean(f))
　　圖 8 中每一個方框表明一個 RDD 分區， T 能夠是任意的類型。經過用戶自定義的過濾函數 f，對每一個數據項操做，將知足條件、返回結果爲 true 的數據項保留。例如，過濾掉 V2 和 V3 保留了 V1，爲區分命名爲 V’1。

 

 

　　圖 8  filter 算子對 RDD 轉換
     

　　（9）distinct

　　distinct將RDD中的元素進行去重操做。圖9中的每一個方框表明一個RDD分區，經過distinct函數，將數據去重。 例如，重複數據V一、 V1去重後只保留一份V1。

 

　　　　圖9  distinct算子對RDD轉換

 

　　（10）subtract

　　subtract至關於進行集合的差操做，RDD 1去除RDD 1和RDD 2交集中的全部元素。圖10中左側的大方框表明兩個RDD，大方框內的小方框表明RDD的分區。 右側大方框
表明合併後的RDD，大方框內的小方框表明分區。 V1在兩個RDD中均有，根據差集運算規則，新RDD不保留，V2在第一個RDD有，第二個RDD沒有，則在新RDD元素中包含V2。
　

　　　　　　　　　　圖10   subtract算子對RDD轉換

 

　　（11） sample

       sample 將 RDD 這個集合內的元素進行採樣，獲取全部元素的子集。用戶能夠設定是否有放回的抽樣、百分比、隨機種子，進而決定採樣方式。內部實現是生成 SampledRDD(withReplacement， fraction， seed)。
　　函數參數設置：
‰ 　　withReplacement=true，表示有放回的抽樣。
‰ 　　withReplacement=false，表示無放回的抽樣。
　　圖 11中 的 每 個 方 框 是 一 個 RDD 分 區。 通 過 sample 函 數， 採 樣 50% 的 數 據。V一、 V二、 U一、 U二、U三、U4 採樣出數據 V1 和 U一、 U2 造成新的 RDD。

     

 

　　　　　　　圖11  sample 算子對 RDD 轉換

 

　　（12）takeSample

　　takeSample（）函數和上面的sample函數是一個原理，可是不使用相對比例採樣，而是按設定的採樣個數進行採樣，同時返回結果再也不是RDD，而是至關於對採樣後的數據進行
Collect（），返回結果的集合爲單機的數組。
　　圖12中左側的方框表明分佈式的各個節點上的分區，右側方框表明單機上返回的結果數組。 經過takeSample對數據採樣，設置爲採樣一份數據，返回結果爲V1。

 

 

　　　　圖12  　　takeSample算子對RDD轉換

 

　　（13） cache

     cache 將 RDD 元素從磁盤緩存到內存。 至關於 persist(MEMORY_ONLY) 函數的功能。
     圖13 中每一個方框表明一個 RDD 分區，左側至關於數據分區都存儲在磁盤，經過 cache 算子將數據緩存在內存。
      

 

　　　　　　圖 13 Cache 算子對 RDD 轉換

 

　　（14） persist

      persist 函數對 RDD 進行緩存操做。數據緩存在哪裏依據 StorageLevel 這個枚舉類型進行肯定。 有如下幾種類型的組合（見10）， DISK 表明磁盤，MEMORY 表明內存， SER 表明數據是否進行序列化存儲。

　　下面爲函數定義， StorageLevel 是枚舉類型，表明存儲模式，用戶能夠經過圖 14-1 按需進行選擇。
　　persist(newLevel:StorageLevel)
　　圖 14-1 中列出persist 函數能夠進行緩存的模式。例如，MEMORY_AND_DISK_SER 表明數據能夠存儲在內存和磁盤，而且以序列化的方式存儲，其餘同理。

 

　　　　　　　　　　　　圖 14-1  persist 算子對 RDD 轉換

　　圖 14-2 中方框表明 RDD 分區。 disk 表明存儲在磁盤， mem 表明存儲在內存。數據最初所有存儲在磁盤，經過 persist(MEMORY_AND_DISK) 將數據緩存到內存，可是有的分區沒法容納在內存，將含有 V一、 V二、 V3 的RDD存儲到磁盤，將含有U1，U2的RDD仍舊存儲在內存。

 

 

 

      圖 14-2   Persist 算子對 RDD 轉換

 

　　（15） mapValues

      mapValues ：針對（Key， Value）型數據中的 Value 進行 Map 操做，而不對 Key 進行處理。

    圖 15 中的方框表明 RDD 分區。 a=>a+2 表明對 (V1,1) 這樣的 Key Value 數據對，數據只對 Value 中的 1 進行加 2 操做，返回結果爲 3。

     

 

　　　　　　圖 15   mapValues 算子 RDD 對轉換

 

　　（16） combineByKey

　　下面代碼爲 combineByKey 函數的定義：
　　combineByKey[C](createCombiner:(V) C,
　　mergeValue:(C, V) C,
　　mergeCombiners:(C, C) C,
　　partitioner:Partitioner,
　　mapSideCombine:Boolean=true,
　　serializer:Serializer=null):RDD[(K,C)]

說明：
‰ 　　createCombiner： V => C， C 不存在的狀況下，好比經過 V 建立 seq C。
‰　　 mergeValue： (C， V) => C，當 C 已經存在的狀況下，須要 merge，好比把 item V
加到 seq C 中，或者疊加。
　　 mergeCombiners： (C， C) => C，合併兩個 C。
‰ 　　partitioner： Partitioner, Shuff le 時須要的 Partitioner。
‰ 　　mapSideCombine ： Boolean = true，爲了減少傳輸量，不少 combine 能夠在 map
端先作，好比疊加，能夠先在一個 partition 中把全部相同的 key 的 value 疊加，
再 shuff le。
‰ 　　serializerClass： String = null，傳輸須要序列化，用戶能夠自定義序列化類：

　　例如，至關於將元素爲 (Int， Int) 的 RDD 轉變爲了 (Int， Seq[Int]) 類型元素的 RDD。圖 16中的方框表明 RDD 分區。如圖，經過 combineByKey， 將 (V1,2)， (V1,1)數據合併爲（ V1,Seq(2,1)）。
　　

 

　　　　　　圖 16  comBineByKey 算子對 RDD 轉換

 

　　（17） reduceByKey

     reduceByKey 是比 combineByKey 更簡單的一種狀況，只是兩個值合併成一個值，（ Int， Int V）to （Int， Int C），好比疊加。因此 createCombiner reduceBykey 很簡單，就是直接返回 v，而 mergeValue和 mergeCombiners 邏輯是相同的，沒有區別。
    函數實現：
    def reduceByKey(partitioner: Partitioner, func: (V, V) => V): RDD[(K, V)]
= {
combineByKey[V]((v: V) => v, func, func, partitioner)
}
　　圖17中的方框表明 RDD 分區。經過用戶自定義函數 (A,B) => (A + B) 函數，將相同 key 的數據 (V1,2) 和 (V1,1) 的 value 相加運算，結果爲（ V1,3）。
     

 

　　　　　　　　圖 17 reduceByKey 算子對 RDD 轉換

 

　　（18）partitionBy

　　partitionBy函數對RDD進行分區操做。
　　函數定義以下。
　　partitionBy（partitioner：Partitioner）
　　若是原有RDD的分區器和現有分區器（partitioner）一致，則不重分區，若是不一致，則至關於根據分區器生成一個新的ShuffledRDD。
　　圖18中的方框表明RDD分區。 經過新的分區策略將原來在不一樣分區的V一、 V2數據都合併到了一個分區。

 

 

　　　　圖18　　partitionBy算子對RDD轉換

 

 （19）Cogroup

 　　cogroup函數將兩個RDD進行協同劃分，cogroup函數的定義以下。
　　cogroup[W]（other： RDD[（K， W）]， numPartitions： Int）： RDD[（K， （Iterable[V]， Iterable[W]））]
　　對在兩個RDD中的Key-Value類型的元素，每一個RDD相同Key的元素分別聚合爲一個集合，而且返回兩個RDD中對應Key的元素集合的迭代器。
　　（K， （Iterable[V]， Iterable[W]））
　　其中，Key和Value，Value是兩個RDD下相同Key的兩個數據集合的迭代器所構成的元組。
　　圖19中的大方框表明RDD，大方框內的小方框表明RDD中的分區。 將RDD1中的數據（U1，1）、 （U1，2）和RDD2中的數據（U1，2）合併爲（U1，（（1，2），（2）））。

 

 

　　　　　　　　圖19  Cogroup算子對RDD轉換

 

 　　（20） join

       join 對兩個須要鏈接的 RDD 進行 cogroup函數操做，將相同 key 的數據可以放到一個分區，在 cogroup 操做以後造成的新 RDD 對每一個key 下的元素進行笛卡爾積的操做，返回的結果再展平，對應 key 下的全部元組造成一個集合。最後返回 RDD[(K， (V， W))]。
　　下 面 代 碼 爲 join 的 函 數 實 現， 本 質 是通 過 cogroup 算 子 先 進 行 協 同 劃 分， 再 通 過flatMapValues 將合併的數據打散。
       this.cogroup(other,partitioner).f latMapValues{case(vs,ws) => for(v<-vs;w<-ws)yield(v,w) }
圖 20是對兩個 RDD 的 join 操做示意圖。大方框表明 RDD，小方框表明 RDD 中的分區。函數對相同 key 的元素，如 V1 爲 key 作鏈接後結果爲 (V1,(1,1)) 和 (V1,(1,2))。

 

　　　　　　　　　　　　　　　　　　　　圖 20   join 算子對 RDD 轉換

 

　　（21）eftOutJoin和rightOutJoin

　　LeftOutJoin（左外鏈接）和RightOutJoin（右外鏈接）至關於在join的基礎上先判斷一側的RDD元素是否爲空，若是爲空，則填充爲空。 若是不爲空，則將數據進行鏈接運算，並
返回結果。
下面代碼是leftOutJoin的實現。
if （ws.isEmpty） {
vs.map（v => （v， None））
} else {
for （v <- vs； w <- ws） yield （v， Some（w））
}

 

2. Actions 算子

　　本質上在 Action 算子中經過 SparkContext 進行了提交做業的 runJob 操做，觸發了RDD DAG 的執行。
例如， Action 算子 collect 函數的代碼以下，感興趣的讀者能夠順着這個入口進行源碼剖析：

/**
* Return an array that contains all of the elements in this RDD.
*/
def collect(): Array[T] = {
/* 提交 Job*/
val results = sc.runJob(this, (iter: Iterator[T]) => iter.toArray)
Array.concat(results: _*)
}

　　（22） foreach

　　foreach 對 RDD 中的每一個元素都應用 f 函數操做，不返回 RDD 和 Array， 而是返回Uint。圖22表示 foreach 算子經過用戶自定義函數對每一個數據項進行操做。本例中自定義函數爲 println()，控制檯打印全部數據項。
　　

 

　　　　　　圖 22 foreach 算子對 RDD 轉換

 

　　（23） saveAsTextFile

　　函數將數據輸出，存儲到 HDFS 的指定目錄。

下面爲 saveAsTextFile 函數的內部實現，其內部
　　經過調用 saveAsHadoopFile 進行實現：
this.map(x => (NullWritable.get(), new Text(x.toString))).saveAsHadoopFile[TextOutputFormat[NullWritable, Text]](path)
將 RDD 中的每一個元素映射轉變爲 (null， x.toString)，而後再將其寫入 HDFS。
　　圖 23中左側方框表明 RDD 分區，右側方框表明 HDFS 的 Block。經過函數將RDD 的每一個分區存儲爲 HDFS 中的一個 Block。

　　

 

　　　　　　　　　　　　圖 23   saveAsHadoopFile 算子對 RDD 轉換

 

  （24）saveAsObjectFile

　　saveAsObjectFile將分區中的每10個元素組成一個Array，而後將這個Array序列化，映射爲（Null，BytesWritable（Y））的元素，寫入HDFS爲SequenceFile的格式。
　　下面代碼爲函數內部實現。
　　map（x=>（NullWritable.get（），new BytesWritable（Utils.serialize（x））））
　　圖24中的左側方框表明RDD分區，右側方框表明HDFS的Block。 經過函數將RDD的每一個分區存儲爲HDFS上的一個Block。

 

 

　　　　　　　　　　　　圖24 saveAsObjectFile算子對RDD轉換

 

 （25） collect

　　collect 至關於 toArray， toArray 已通過時不推薦使用， collect 將分佈式的 RDD 返回爲一個單機的 scala Array 數組。在這個數組上運用 scala 的函數式操做。
　　圖 25中左側方框表明 RDD 分區，右側方框表明單機內存中的數組。經過函數操做，將結果返回到 Driver 程序所在的節點，以數組形式存儲。

 

 

　　圖 25   Collect 算子對 RDD 轉換　

 

　　（26）collectAsMap

　　collectAsMap對（K，V）型的RDD數據返回一個單機HashMap。 對於重複K的RDD元素，後面的元素覆蓋前面的元素。
　　圖26中的左側方框表明RDD分區，右側方框表明單機數組。 數據經過collectAsMap函數返回給Driver程序計算結果，結果以HashMap形式存儲。

 

 

 

　　　　　　　　　　圖26 CollectAsMap算子對RDD轉換

 

 　　（27）reduceByKeyLocally

　　實現的是先reduce再collectAsMap的功能，先對RDD的總體進行reduce操做，而後再收集全部結果返回爲一個HashMap。

 

 　　（28）lookup

下面代碼爲lookup的聲明。
lookup（key：K）：Seq[V]
Lookup函數對（Key，Value）型的RDD操做，返回指定Key對應的元素造成的Seq。 這個函數處理優化的部分在於，若是這個RDD包含分區器，則只會對應處理K所在的分區，而後返回由（K，V）造成的Seq。 若是RDD不包含分區器，則須要對全RDD元素進行暴力掃描處理，搜索指定K對應的元素。
　　圖28中的左側方框表明RDD分區，右側方框表明Seq，最後結果返回到Driver所在節點的應用中。

 

 

　　　　　　圖28  lookup對RDD轉換

 

　　（29） count

 

　　count 返回整個 RDD 的元素個數。
　　內部函數實現爲：
　　defcount():Long=sc.runJob(this,Utils.getIteratorSize_).sum
　　圖 29中，返回數據的個數爲 5。一個方塊表明一個 RDD 分區。

 

　　　　 圖29 count 對 RDD 算子轉換

 

　　（30）top

top可返回最大的k個元素。 函數定義以下。
top（num：Int）（implicit ord：Ordering[T]）：Array[T]

相近函數說明以下。
·top返回最大的k個元素。
·take返回最小的k個元素。
·takeOrdered返回最小的k個元素，而且在返回的數組中保持元素的順序。
·first至關於top（1）返回整個RDD中的前k個元素，能夠定義排序的方式Ordering[T]。
返回的是一個含前k個元素的數組。

 

　　（31）reduce

　　reduce函數至關於對RDD中的元素進行reduceLeft函數的操做。 函數實現以下。
　　Some（iter.reduceLeft（cleanF））
　　reduceLeft先對兩個元素<K，V>進行reduce函數操做，而後將結果和迭代器取出的下一個元素<k，V>進行reduce函數操做，直到迭代器遍歷完全部元素，獲得最後結果。在RDD中，先對每一個分區中的全部元素<K，V>的集合分別進行reduceLeft。 每一個分區造成的結果至關於一個元素<K，V>，再對這個結果集合進行reduceleft操做。
　　例如：用戶自定義函數以下。
　　f：（A，B）=>（A._1+」@」+B._1，A._2+B._2）
　　圖31中的方框表明一個RDD分區，經過用戶自定函數f將數據進行reduce運算。 示例
最後的返回結果爲V1@[1]V2U！@U2@U3@U4，12。

 

 

 

圖31 reduce算子對RDD轉換

 

　　（32）fold

　　fold和reduce的原理相同，可是與reduce不一樣，至關於每一個reduce時，迭代器取的第一個元素是zeroValue。
　　圖32中經過下面的用戶自定義函數進行fold運算，圖中的一個方框表明一個RDD分區。 讀者能夠參照reduce函數理解。
　　fold（（」V0@」，2））（ （A，B）=>（A._1+」@」+B._1，A._2+B._2））

 

 

　　　　　　　　　　圖32  fold算子對RDD轉換

 

 　　（33）aggregate

 　　aggregate先對每一個分區的全部元素進行aggregate操做，再對分區的結果進行fold操做。
　　aggreagate與fold和reduce的不一樣之處在於，aggregate至關於採用歸併的方式進行數據彙集，這種彙集是並行化的。 而在fold和reduce函數的運算過程當中，每一個分區中須要進行串行處理，每一個分區串行計算完結果，結果再按以前的方式進行彙集，並返回最終彙集結果。
　　函數的定義以下。
aggregate[B]（z： B）（seqop： （B，A） => B，combop： （B，B） => B）： B
　　圖33經過用戶自定義函數對RDD 進行aggregate的彙集操做，圖中的每一個方框表明一個RDD分區。
　　rdd.aggregate（」V0@」，2）（（A，B）=>（A._1+」@」+B._1，A._2+B._2）），（A，B）=>（A._1+」@」+B_1，A._@+B_.2））
　　最後，介紹兩個計算模型中的兩個特殊變量。
　　廣播（broadcast）變量：其普遍用於廣播Map Side Join中的小表，以及廣播大變量等場景。 這些數據集合在單節點內存可以容納，不須要像RDD那樣在節點之間打散存儲。
Spark運行時把廣播變量數據發到各個節點，並保存下來，後續計算能夠複用。 相比Hadoo的distributed cache，廣播的內容能夠跨做業共享。 Broadcast的底層實現採用了BT機制。

 

　　　　　　　　圖33  aggregate算子對RDD轉換

　　②表明V。　　③表明U。　　accumulator變量：容許作全局累加操做，如accumulator變量普遍使用在應用中記錄當前的運行指標的情景。--------------------- 做者：魅影獵鷹 來源：CSDN 原文：https://blog.csdn.net/qq_32595075/article/details/79918644 版權聲明：本文爲博主原創文章，轉載請附上博文連接！