spark RDD算子中Actions算子怎么用

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

Actions算子

本質上在Actions算子中通過SparkContext執行提交作業的runJob操作，觸發了RDD DAG的執行。

1．無輸出

（1）foreach(f)

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

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

2．HDFS

saveAsTextFile(path, compressionCodecClass=None)

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

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

3．Scala集合和數據類型

（1）collect()

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

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

（2）collectAsMap()

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

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

（3）reduceByKeyLocally(func)

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

（4）lookup(key)

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

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

（5）count()

count返回整個RDD的元素個數。內部函數實現如下。
在圖3-31中，返回數據的個數為5。一個方塊代表一個RDD分區。

（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（接受兩個參數，返回一個參數）聚集數據集中的所有元素。這個功能必須可交換且可關聯的，從而可以正確的被并行執行。

例子：

>>> 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算子怎么用”這篇文章的所有內容，感謝各位的閱讀！相信大家都有了一定的了解，希望分享的內容對大家有所幫助，如果還想學習更多知識，歡迎關注億速云行業資訊頻道！