spark大資料分析:spark core(4) 轉換算子

partitionBy

reduceByKey

groupByKey

combineByKey

aggregateByKey

foldByKey

sortByKey

join

cogroup

mapValues

flatMapValues

map

flatMap

filter

distinct

mapPartition

mapPartitionWithIndex

union

insertsection

subtract

coalesce

repartition

randomSplit

glom

zip

zipPartitions

zipWithIndex

zipWithUniqueld

sortBy

基礎轉換

鍵值對轉換

源碼

一進一出

案例

基于map以及reduceByKey的封裝,同時重載方法可以修改分區數,預設情況下distinct和原分區數一緻

與map不同的是,某個RDD有2個分區,10個元素,那麼map是将元素直接依次應用到f函數上,而mapPartition操作中則是先周遊分區,再周遊分區中的元素,第一個參數是疊代器,第二個參數是是否保留分區資訊,預設是false,除非上遊是簡直對且函數對key進行修改,否則保留分區資訊無意義

與mapPartition 不同,輸出結果帶分區号

傳回資料并集

1,2,3,4,5,6,7,8,9,10,11,12,13,14,15,16,17,18,19,20,15,16,17,18,19,20,21,22,23,24,25,26,27,28,29,30,31,32,33,34,35,36,37,38,39,40,41,42,43,44,45,46,47,48,49,50,51,52,53,54,55,56,57,58,59,60,61,62,63,64,65,66,67,68,69,70,71,72,73,74,75,76,77,78,79,80,81,82,83,84,85,86,87,88,89,90,91,92,93,94,95,96,97,98,99,100

不會對資料去重

傳回資料交集

15,16,17,18,19,20

傳回資料差集

12,1,13,2,14,3,4,5,6,7,8,9,10,11

減少分區

第二個參數預設是false,如果目标分區數大于原始分區,此操作無意義,維持原分區不變,對于較多分區向較少分區轉換,coalesce 不會産生shuffle,而向多分區轉換類比 repartition

其實底層是 coalesce 但是shuffle 參數是true,是以無論是多分區向少分區轉換還是少分區向多分區轉換都會産生shuffle,coalesce 開啟shuffle 與 repartition 一緻

拆分RDD配置設定權重值

最好是 1,3,4,2 之間加起比例為1 1/10 + 3/10 + 4/10 + 2/10 = 1

每個RDD 中元素個數近似為權值

将RDD中每個分區變成一個數組,放置在新的RDD中

将兩個RDD以鍵值對方式進行合并,key為EDD第一個元素,value為RDD 第二個元素

如果兩個RDD分區數不一緻,報錯

将兩個RDD按照分區号進行合并

将RDD中元素與該RDD 索引進行合并

該元素與元素的唯一id對應

id生成算法

xn +k

n分區數,k 分區号,x 自然數 從0開始

排序,預設升序

重分區生成shuffleRDD

對于RDD中相同的K和V的值進行聚合

設定RDD分區數,指定分區數個數

(aads,321) (zs,45)

createCombiner:

該參數是一個操作,當周遊RDD中每一個元素,如果該元素對于的K是被第一次通路,則會調用createCombiner操作來建立組合器進而将元素的值從V類轉換為C類

mergeValue

将 C 類型和V 類型轉換為C類型

combineByKeyWithClassTag

一個RDD 有多個分區,多個擁有相同的k 可以被分散到不同的分區中,每個分區有獨立的Combiner,擁有相同的鍵的元素會先在各個所在分區進行聚合,然後通過 mergeCombiners 函數将多個分區中的函數将多個分區各自的聚合結果按照鍵做最後一次聚合

(zs,12.0),(aads,277.0)

zeroValue

設定聚合時的初始值,可以是數字或者集合

seqOp

将值V聚合到類型為U的對象中

combOp

跨分區聚合,對資料進行最終的彙總時調用此操作

原理與combineByKey相同,是對combinerByKey的封裝,簡化了操作 aggregateByKey

和reduceBykey雷同,但是接受傳入初始值根據key聚合

zeroValue 為初始值

結果

(zs,2122)->(aads,2321f233f)

aggregateByKey , foldByKey , reduceByKey 三個操作本質都是調用combineByKey

combineByKey 和 aggregateByKey 輸入輸出類型可以不一緻,但是 foldByKey 與 reduceByKey 輸入輸出類型必須一緻

根據key排序,可以選擇倒序或者正序

(aads,321f),(aads,233f),(zs,1),(zs,2)

相同的key的值聚合在一起構成一個集合,不同的舍棄

代碼

(zs,(1,2)),(zs,(2,2)),(aads,(321f,233f)),(aads,(233f,233f))

相當于求并集

(aads2,(CompactBuffer(),CompactBuffer(321f))),(zs,(CompactBuffer(1, 2),CompactBuffer(2))),(zs1,(CompactBuffer(),CompactBuffer(1))),(aads,(CompactBuffer(321f, 233f),CompactBuffer(233f)))

對v進行操作,傳回新的RDD

(zs,1 元),(zs,2 元),(aads,321f 元),(aads,233f 元)

(zs,1),(zs,),(zs,2),(zs, ),(zs,元),(zs,2),(zs,),(zs,3),(zs, ),(zs,元),(aads,3),(aads,2),(aads,1),(aads,f),(aads,),(aads,4),(aads, ),(aads,元),(aads,2),(aads,3),(aads,3),(aads,f),(aads, ),(aads,元)