當前位置：首頁 > 编程资源 > 编程问答 >内容正文

编程问答

Spark笔记：复杂RDD的API的理解（上）

發布時間：2024/1/17 编程问答 21 豆豆

生活随笔收集整理的這篇文章主要介紹了 Spark笔记：复杂RDD的API的理解（上）小編覺得挺不錯的,現在分享給大家,幫大家做個參考.

本篇接著講解RDD的API，講解那些不是很容易理解的API，同時本篇文章還將展示如何將外部的函數引入到RDD的API里使用，最后通過對RDD的API深入學習，我們還講講一些和RDD開發相關的scala語法。

1) ?aggregate(zeroValue)(seqOp,combOp)

　 ?該函數的功能和reduce函數一樣，也是對數據進行聚合操作，不過aggregate可以返回和原RDD不同的數據類型，使用時候還要提供初始值。

　　我們來看看下面的用法，代碼如下：

val?rddInt:?RDD[Int]?=?sc.parallelize(List(1,?2,?3,?4,?5),?1)

val?rddAggr1:?(Int, Int)?=?rddInt.aggregate((0,?0))((x, y)?=> (x._1?+ y, x._2?+?1), (x, y)?=> (x._1?+ y._1, x._2?+ y._2))

println("====aggregate 1====:"?+ rddAggr1.toString())?// (15,5)

　　該方法是將有數字組成的RDD的數值進行求和，同時還要統計元素的個數，這樣我們就可以計算出一個平均值，這點在實際運算中是非常有用的。

　　假如讀者不太懂scala語言，或者就算懂那么一點點scala語法，該API的使用還是讓人很難理解的，這個x是什么東西，這個y又是什么東西，為什么它們放在一起這么運算就可以得到預期結果呢？

　　其實aggregate方法使用了scala里元組的結構，元組是scala里很具特色的數據結構，我們看看下面的代碼：

val?tuple2Param1:Tuple2[String,Int]?=?Tuple2("x01",12)// 標準定義二元組

val?tuple2Param2:(String,Int)?=?("x02",29)// 字面量定義二元組

/* 結果: x01:12*/

println("====tuple2Param1====:"?+ tuple2Param1._1?+?":"?+ tuple2Param1._2)

/* 結果: x02:29 */

println("====tuple2Param2====:"?+ tuple2Param2._1?+?":"?+ tuple2Param2._2)

val?tuple6Param1:Tuple6[String,Int,Int,Int,Int,String]?=?Tuple6("xx01",1,2,3,4,"x1x")// 標準定義6元組

val?tuple6Param2:(String,Int,Int,Int,Int,String)?=?("xx02",1,2,3,4,"x2x")// 字面量定義6元組

/* 結果: xx01:1:2:3:4:x1x */

println("====tuple6Param1====:"?+ tuple6Param1._1?+?":"?+ tuple6Param1._2?+?":"?+ tuple6Param1._3?+?":"?+ tuple6Param1._4?+?":"?+ tuple6Param1._5?+?":"?+ tuple6Param1._6)

/* 結果: xx02:1:2:3:4:x2x */

println("====tuple6Param2====:"?+ tuple6Param2._1?+?":"?+ tuple6Param2._2?+?":"?+ tuple6Param2._3?+?":"?+ tuple6Param2._4?+?":"?+ tuple6Param2._5?+?":"?+ tuple6Param2._6)

　　元組在scala里使用Tuple來構造，不過實際運用中我們會給Tuple帶上數字后綴，例如Tuple2就是二元組它包含兩個元素，Tuple6是6元組它包含6個元素，元組看起來很像數組，但是數組只能存儲相同數據類型的數據結構，而元組是可以存儲不同數據類型的數據結構，元組里元素訪問使用_1,_2這樣的形式，第一個元素是從1開始標記的，這點和數組是不同的。實際使用中我們很少使用Tuple構造元組，而是使用字面量定義方式（參見代碼注釋），由此我們可以看出spark里鍵值對RDD其實就是使用二元組來表示鍵值對數據結構，回到aggregate方法，它的運算也是通過二元組這種數據結構完成的。

　　下面我們來看看aggregate的運算過程，這里我將aggregate方法里的算子都使用外部函數，代碼如下所示：

def?aggrFtnOne(par:?((Int, Int), Int)):?(Int, Int)?=?{

??/*

?????*aggregate的初始值為(0,0):

??????====aggrFtnOne Param===:((0,0),1)

??====aggrFtnOne Param===:((1,1),2)

??====aggrFtnOne Param===:((3,2),3)

??====aggrFtnOne Param===:((6,3),4)

??====aggrFtnOne Param===:((10,4),5)*/

??/*

?????*aggregate的初始值為(1,1):

??????====aggrFtnOne Param===:((1,1),1)

??????====aggrFtnOne Param===:((2,2),2)

??????====aggrFtnOne Param===:((4,3),3)

??????====aggrFtnOne Param===:((7,4),4)

??????====aggrFtnOne Param===:((11,5),5)

?????* */

??println("====aggrFtnOne Param===:"?+ par.toString())

??val?ret:?(Int, Int)?=?(par._1._1?+ par._2, par._1._2?+?1)

??ret

}

def?aggrFtnTwo(par:?((Int, Int), (Int, Int))):?(Int, Int)?=?{

??/*aggregate的初始值為(0,0):::::((0,0),(15,5))*/

??/*aggregate的初始值為(1,1):::::((1,1),(16,6))*/

??println("====aggrFtnTwo Param===:"?+ par.toString())

??val?ret:?(Int, Int)?=?(par._1._1?+ par._2._1, par._1._2?+ par._2._2)

??ret

}

??val?rddAggr2:?(Int, Int)?=?rddInt.aggregate((0,?0))((x, y)?=> aggrFtnOne(x, y), (x, y)?=> aggrFtnTwo(x, y))?// 參數可以省略元組的括號

??println("====aggregate 2====:"?+ rddAggr2.toString())?// (15,5)

??val?rddAggr3:?(Int, Int)?=?rddInt.aggregate((1,?1))((x, y)?=> aggrFtnOne((x, y)), (x, y)?=> aggrFtnTwo((x, y)))?// 參數使用元組的括號

??println("====aggregate 3====:"?+ rddAggr3.toString())?// (17,7)

　　由以上代碼我們就可以清晰看到aggregate方法的實際運算過程了。

　　aggrFtnOne方法的參數格式是((Int, Int), Int)，這個復合二元組里第二個元素才是實際的值，而第一個元素就是我們給出的初始化值，第一個元素里的第一個值就是我們實際求和的值，里面第二個元素就是累計記錄元素個數的值。

　　aggrFtnTwo方法的參數里的二元組第一個元素還是初始化值，第二個元素則是aggrFtnOne計算的結果，這樣我們就可以計算出我們要的結果。

　　作為對比我將初始化參數改為(1,1)二元組，最終結果在求和運算以及計算元素個數上都會加2，這是因為初始化值兩次參入求和所致的，由上面代碼我們可以很清晰的看到原因所在。

　　如果我們想要結果二元組里第一個元素求積那么初始化值就不能是(0,0)，而應該是(1,0),理解了原理我們就很清晰知道初始值該如何設定了，具體代碼如下：

1 2	val?rddAggr4:?(Int, Int)?=?rddInt.aggregate((1,?0))((x, y)?=> (x._1?* y, x._2?+?1), (x, y)?=> (x._1?* y._1, x._2?+ y._2)) println("====aggregate 4====:"?+ rddAggr4.toString())?// (120,5)

2) fold(zero)(func)

　該函數和reduce函數功能一樣，只不過使用時候需要加上初始化值。

　代碼如下所示：

def?foldFtn(par:?(Int, Int)):?Int?=?{

??/*fold初始值為0：

??????=====foldFtn Param====:(0,1)

??????=====foldFtn Param====:(1,2)

??????=====foldFtn Param====:(3,3)

??????=====foldFtn Param====:(6,4)

??????=====foldFtn Param====:(10,5)

??????=====foldFtn Param====:(0,15)

?????* */

??/*

?????* fold初始值為1:

??????=====foldFtn Param====:(1,1)

??????=====foldFtn Param====:(2,2)

??????=====foldFtn Param====:(4,3)

??????=====foldFtn Param====:(7,4)

??????=====foldFtn Param====:(11,5)

??????=====foldFtn Param====:(1,16)

?????* */

??println("=====foldFtn Param====:"?+ par.toString())

??val?ret:?Int?=?par._1?+ par._2

??ret

}

??val?rddFold2:?Int?=?rddInt.fold(0)((x, y)?=> foldFtn(x, y))?// 參數可以省略元組的括號

??println("====fold 2=====:"?+ rddFold2)?// 15

??val?rddFold3:?Int?=?rddInt.fold(1)((x, y)?=> foldFtn((x, y)))?// 參數使用元組的括號

??println("====fold 3====:"?+ rddFold3)?// 17

　　我們發現當初始化值為1時候，求和增加的不是1而是2，原因就是fold計算時候為了湊齊一個完整的二元組，在第一個元素計算以及最后一個元素計算時候都會讓初始化值湊數組成二元組，因此初始值會被使用兩遍求和，因此實際結果就不是增加1，而是增加2了。

　　作為對比我們看看reduce實際運算過程，代碼如下：

def?reduceFtn(par:(Int,Int)):Int?=?{

??/*

???* ======reduceFtn Param=====:1:2

???????????======reduceFtn Param=====:3:3

?????======reduceFtn Param=====:6:4

?????======reduceFtn Param=====:10:5

???*/

??println("======reduceFtn Param=====:"?+ par._1?+?":"?+ par._2)

??par._1?+ par._2

}

??val?rddReduce1:Int?=?rddInt.reduce((x,y)?=> x + y)

??println("====rddReduce 1====:"?+ rddReduce1)// 15

???

??val?rddReduce2:Int?=?rddInt.reduce((x,y)?=> reduceFtn(x,y))

??println("====rddReduce 2====:"?+ rddReduce2)// 15

3) combineByKey[C](createCombiner: Int => C, mergeValue: (C, Int) => C, mergeCombiners: (C, C) => C): RDD[(String, C)]

　　combineByKey作用是使用不同的返回類型合并具有相同鍵的值，combineByKey適用鍵值對RDD，普通RDD是沒有這個方法。

　　有上面定義我們看到combineByKey會經過三輪運算，前一個運算步驟結果就是下一個運算步驟的參數，我們看下面的代碼：

def?combineFtnOne(par:Int):(Int,Int)?=?{

??/*

???* ====combineFtnOne Param====:2

?????====combineFtnOne Param====:5

?????====combineFtnOne Param====:8

?????====combineFtnOne Param====:3

???*/

??println("====combineFtnOne Param====:"?+ par)

??val?ret:(Int,Int)?=?(par,1)

??ret

}

def?combineFtnTwo(par:((Int,Int),Int)):(Int,Int)?=?{

??/*

????====combineFtnTwo Param====:((2,1),12)

????====combineFtnTwo Param====:((8,1),9)

???* */

??println("====combineFtnTwo Param====:"?+ par.toString())

??val?ret:(Int,Int)?=?(par._1._1?+ par._2,par._1._2?+?1)

??ret

}

def?combineFtnThree(par:((Int,Int),(Int,Int))):(Int,Int)?=?{

??/*

???* 無結果打印

???*/

??println("@@@@@@@@@@@@@@@@@@")

??println("====combineFtnThree Param===:"?+ par.toString())

??val?ret:(Int,Int)?=?(par._1._1?+ par._2._1,par._1._2?+ par._2._2)

??ret

}

??val?rddPair:?RDD[(String, Int)]?=?sc.parallelize(List(("x01",?2), ("x02",?5), ("x03",?8), ("x04",?3), ("x01",?12), ("x03",?9)),?1)

???

??/* def combineByKey[C](createCombiner: Int => C, mergeValue: (C, Int) => C, mergeCombiners: (C, C) => C): RDD[(String, C)] */???

??val?rddCombine1:RDD[(String,(Int,Int))]?=?rddPair.combineByKey(x?=> (x,?1), (com:?(Int, Int), x)?=> (com._1?+ x, com._2?+?1), (com1:?(Int, Int), com2:?(Int, Int))?=> (com1._1?+ com2._1, com1._2?+ com2._2))

??println("====combineByKey 1====:"?+ rddCombine1.collect().mkString(","))?// (x02,(5,1)),(x03,(17,2)),(x01,(14,2)),(x04,(3,1))

???

??val?rddCombine2:RDD[(String,(Int,Int))]?=?rddPair.combineByKey(x?=> combineFtnOne(x), (com:?(Int, Int), x)?=> combineFtnTwo(com,x), (com1:?(Int, Int), com2:?(Int, Int))?=> combineFtnThree(com1,com2))

??println("=====combineByKey 2====:"?+ rddCombine2.collect().mkString(","))?// (x02,(5,1)),(x03,(17,2)),(x01,(14,2)),(x04,(3,1))

　　這個算法和上面aggregate求和方法很像，不過combineByKey很奇怪，它第三個算子似乎并沒有被執行，第二個算子打印的信息也不齊備，不過我認為它們都執行了，只不過有些語句沒有打印出來，至于原因為何，我以后再研究下吧。

創作挑戰賽新人創作獎勵來咯，堅持創作打卡瓜分現金大獎

總結

以上是生活随笔為你收集整理的Spark笔记：复杂RDD的API的理解（上）的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。

上一篇： Spark笔记：RDD基本操作（上）
下一篇： Spark笔记：复杂RDD的API的理解