温馨提示×

温馨提示×

您好,登录后才能下订单哦!

密码登录×
登录注册×
其他方式登录
点击 登录注册 即表示同意《亿速云用户服务条款》

spark RDD算子中Actions算子怎么用

发布时间:2021-12-10 13:35:45 来源:亿速云 阅读:203 作者:小新 栏目:云计算

这篇文章主要为大家展示了“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(),控制台打印所有数据项。

spark RDD算子中Actions算子怎么用

2.HDFS

saveAsTextFile(path, compressionCodecClass=None)

函数将数据输出,存储到HDFS的指定目录。
将RDD中的每个元素映射转变为(Null, x.toString),然后再将其写入HDFS。

图3-26中左侧的方框代表RDD分区,右侧方框代表HDFS的Block。通过函数将RDD的每个分区存储为HDFS中的一个Block。

spark RDD算子中Actions算子怎么用

3.Scala集合和数据类型

(1)collect()

collect将分布式的RDD返回为一个单机的scala Array数组。在这个数组上运用scala的函数式操作。

图3-28中的左侧方框代表RDD分区,右侧方框代表单机内存中的数组。通过函数操作,将结果返回到Driver程序所在的节点,以数组形式存储。

spark RDD算子中Actions算子怎么用

(2)collectAsMap()

collectAsMap对(K, V)型的RDD数据返回一个单机HashMap。对于重复K的RDD元素,后面的元素覆盖前面的元素。

图3-29中的左侧方框代表RDD分区,右侧方框代表单机数组。数据通过collectAsMap函数返回给Driver程序计算结果,结果以HashMap形式存储。

spark RDD算子中Actions算子怎么用

(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所在节点的应用中。

spark RDD算子中Actions算子怎么用

(5)count()

count返回整个RDD的元素个数。内部函数实现如下。
在图3-31中,返回数据的个数为5。一个方块代表一个RDD分区。

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(接受两个参数,返回一个参数)聚集数据集中的所有元素。这个功能必须可交换且可关联的,从而可以正确的被并行执行。

例子:

>>> 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算子怎么用”这篇文章的所有内容,感谢各位的阅读!相信大家都有了一定的了解,希望分享的内容对大家有所帮助,如果还想学习更多知识,欢迎关注亿速云行业资讯频道!

向AI问一下细节

免责声明:本站发布的内容(图片、视频和文字)以原创、转载和分享为主,文章观点不代表本网站立场,如果涉及侵权请联系站长邮箱:is@yisu.com进行举报,并提供相关证据,一经查实,将立刻删除涉嫌侵权内容。

AI