溫馨提示×

溫馨提示×

您好,登錄后才能下訂單哦!

密碼登錄×
登錄注冊×
其他方式登錄
點擊 登錄注冊 即表示同意《億速云用戶服務條款》

spark RDD算子中Actions算子怎么用

發(fā)布時間:2021-12-10 13:35:45 來源:億速云 閱讀:193 作者:小新 欄目:云計算

這篇文章主要為大家展示了“spark RDD算子中Actions算子怎么用”,內容簡而易懂,條理清晰,希望能夠幫助大家解決疑惑,下面讓小編帶領大家一起研究并學習一下“spark RDD算子中Actions算子怎么用”這篇文章吧。

Actions算子

本質上在Actions算子中通過SparkContext執(zhí)行提交作業(yè)的runJob操作,觸發(fā)了RDD DAG的執(zhí)行。

1.無輸出

(1)foreach(f)

對RDD中的每個元素都應用f函數操作,不返回RDD和Array,而是返回Uint。

圖3-25表示foreach算子通過用戶自定義函數對每個數據項進行操作。本例中自定義函數為println(),控制臺打印所有數據項。

spark RDD算子中Actions算子怎么用

2.HDFS

saveAsTextFile(path, compressionCodecClass=None)

函數將數據輸出,存儲到HDFS的指定目錄。
將RDD中的每個元素映射轉變?yōu)?Null, x.toString),然后再將其寫入HDFS。

圖3-26中左側的方框代表RDD分區(qū),右側方框代表HDFS的Block。通過函數將RDD的每個分區(qū)存儲為HDFS中的一個Block。

spark RDD算子中Actions算子怎么用

3.Scala集合和數據類型

(1)collect()

collect將分布式的RDD返回為一個單機的scala Array數組。在這個數組上運用scala的函數式操作。

圖3-28中的左側方框代表RDD分區(qū),右側方框代表單機內存中的數組。通過函數操作,將結果返回到Driver程序所在的節(jié)點,以數組形式存儲。

spark RDD算子中Actions算子怎么用

(2)collectAsMap()

collectAsMap對(K, V)型的RDD數據返回一個單機HashMap。對于重復K的RDD元素,后面的元素覆蓋前面的元素。

圖3-29中的左側方框代表RDD分區(qū),右側方框代表單機數組。數據通過collectAsMap函數返回給Driver程序計算結果,結果以HashMap形式存儲。

spark RDD算子中Actions算子怎么用

(3)reduceByKeyLocally(func)

實現(xiàn)的是先reduce再collectAsMap的功能,先對RDD的整體進行reduce操作,然后再收集所有結果返回為一個HashMap。

(4)lookup(key)

Lookup函數對(Key, Value)型的RDD操作,返回指定Key對應的元素形成的Seq。這個函數處理優(yōu)化的部分在于,如果這個RDD包含分區(qū)器,則只會對應處理K所在的分區(qū),然后返回由(K, V)形成的Seq。如果RDD不包含分區(qū)器,則需要對全RDD元素進行暴力掃描處理,搜索指定K對應的元素。

圖3-30中的左側方框代表RDD分區(qū),右側方框代表Seq,最后結果返回到Driver所在節(jié)點的應用中。

spark RDD算子中Actions算子怎么用

(5)count()

count返回整個RDD的元素個數。內部函數實現(xiàn)如下。
在圖3-31中,返回數據的個數為5。一個方塊代表一個RDD分區(qū)。

spark RDD算子中Actions算子怎么用

(6)top(num, key=None)

top可返回最大的k個元素。
相近函數說明如下。

top返回最大的k個元素。

take返回最小的k個元素。

takeOrdered返回最小的k個元素,并且在返回的數組中保持元素的順序。

first相當于top(1)返回整個RDD中的前k個元素,可以定義排序的方式Ordering[T]。返回的是一個含前k個元素的數組。

(7)reduce(f)

通過函數func(接受兩個參數,返回一個參數)聚集數據集中的所有元素。這個功能必須可交換且可關聯(lián)的,從而可以正確的被并行執(zhí)行。

例子:

>>> from operator import add
>>> sc.parallelize([1, 2, 3, 4, 5]).reduce(add)
15
>>> sc.parallelize((2 for _ in range(10))).map(lambda x: 1).cache().reduce(add)
10

(8)fold(zeroValue, op)

fold和reduce的原理相同,但是與reduce不同,相當于每個reduce時,迭代器取的第一個元素是zeroValue。

>>> from operator import add
>>> sc.parallelize([1, 2, 3, 4, 5]).fold(0, add)
15

以上是“spark RDD算子中Actions算子怎么用”這篇文章的所有內容,感謝各位的閱讀!相信大家都有了一定的了解,希望分享的內容對大家有所幫助,如果還想學習更多知識,歡迎關注億速云行業(yè)資訊頻道!

向AI問一下細節(jié)

免責聲明:本站發(fā)布的內容(圖片、視頻和文字)以原創(chuàng)、轉載和分享為主,文章觀點不代表本網站立場,如果涉及侵權請聯(lián)系站長郵箱:is@yisu.com進行舉報,并提供相關證據,一經查實,將立刻刪除涉嫌侵權內容。

AI