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

成都創(chuàng)新互聯(lián)是一家專業(yè)提供大洼企業(yè)網(wǎng)站建設(shè),專注與網(wǎng)站設(shè)計制作、成都網(wǎng)站設(shè)計、H5頁面制作、小程序制作等業(yè)務(wù)。10年已為大洼眾多企業(yè)、政府機(jī)構(gòu)等服務(wù)。創(chuàng)新互聯(lián)專業(yè)的建站公司優(yōu)惠進(jìn)行中。

Actions算子

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

1．無輸出

（1）foreach(f)

對RDD中的每個元素都應(yīng)用f函數(shù)操作，不返回RDD和Array，而是返回Uint。

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

2．HDFS

saveAsTextFile(path, compressionCodecClass=None)

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

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

3．Scala集合和數(shù)據(jù)類型

（1）collect()

collect將分布式的RDD返回為一個單機(jī)的scala Array數(shù)組。在這個數(shù)組上運(yùn)用scala的函數(shù)式操作。

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

（2）collectAsMap()

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

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

（3）reduceByKeyLocally(func)

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

（4）lookup(key)

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

圖3-30中的左側(cè)方框代表RDD分區(qū)，右側(cè)方框代表Seq，最后結(jié)果返回到Driver所在節(jié)點(diǎn)的應(yīng)用中。

（5）count()

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

（6）top(num, key=None)

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

top返回最大的k個元素。

take返回最小的k個元素。

takeOrdered返回最小的k個元素，并且在返回的數(shù)組中保持元素的順序。

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

（7）reduce(f)

通過函數(shù)func（接受兩個參數(shù)，返回一個參數(shù)）聚集數(shù)據(jù)集中的所有元素。這個功能必須可交換且可關(guān)聯(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不同，相當(dāng)于每個reduce時，迭代器取的第一個元素是zeroValue。

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

以上是“spark RDD算子中Actions算子怎么用”這篇文章的所有內(nèi)容，感謝各位的閱讀！相信大家都有了一定的了解，希望分享的內(nèi)容對大家有所幫助，如果還想學(xué)習(xí)更多知識，歡迎關(guān)注創(chuàng)新互聯(lián)行業(yè)資訊頻道！