RDD——Action算子

常用Action算子

• countByKey
• collect
• reduce
• fold
• first
• take
• top
• count
• takeSample
• takeOrdered
• foreach
• saveAsTextFile

1. countByKey()

返回值是一个 字典类型

## 读取文本文件
rdd1 &＃61; sc.textFile(f"file:///{ROOT}/data/input/words.txt")
rdd2 &＃61; rdd1.flatMap(lambda x:x.split(" ")).map(lambda x:(x,1))
print(rdd2.collect())
# [(&＃39;hello&＃39;, 1), (&＃39;spark&＃39;, 1), (&＃39;hello&＃39;, 1), (&＃39;hadoop&＃39;, 1), (&＃39;hello&＃39;, 1), (&＃39;flink&＃39;, 1)]
res &＃61; rdd2.countByKey()
print(res)
# defaultdict(, {&＃39;hello&＃39;: 3, &＃39;spark&＃39;: 1, &＃39;hadoop&＃39;: 1, &＃39;flink&＃39;: 1})
print(type(res))
#


2. collect()

将RDD各分区数据统一收集到Driver 形成一个List对象

确定数据量不是太大 不然会把Driver 内存撑爆

3. reduce()

将RDD数据集 进行聚合&＃xff0c;返回值就是传入的数据同类型

注意reduceByKey 返回值是RDD

rdd &＃61; sc.parallelize([1, 2, 3, 4, 5])
res &＃61; rdd.reduce(lambda a,b:a&＃43;b)
res
# 15


4. fold

初始值 在分区内 和 分区间 都会作用

一般不使用

5. first()

RDD的第一个元素

6. take(N)

取RDD的前N个元素 组合成List 返回

7. top(N)

对RDD数据进行降序排序 &＃xff0c;取得前N个 组合成List

8. count()

计算RDD有多少条数据&＃xff0c;返回值是一个数字

9. takeSample&＃xff08;参数1&＃xff1a;True or False,参数2&＃xff1a;采样数&＃xff0c;参数3&＃xff1a;随机种子&＃xff09;

   参数1&＃xff1a; True 表示可以重复取同一个数据

随机抽样RDD数据 返回List

10. takeOrdered(N&＃xff0c;参数2)

    升序排列

    参数2 可以对排序数据进行改变 不改变本身

对RDD进行排序 取前N个

rdd &＃61; sc.parallelize([1, 3, 2, 4, 7, 9, 6], 1)
print(rdd.takeOrdered(3))
# [1,2,3]
print(rdd.takeOrdered(3, lambda x: -x))
# [9,7,6]


11. foreach(func)

对RDD每一个元素 执行提供的逻辑操作&＃xff08;同map&＃xff09;也不改变传入的RDD 但是没有返回值

rdd &＃61; sc.parallelize([1, 2, 3, 4, 5])
rdd2 &＃61; rdd.foreach(lambda x:print(x*10) )
# print(rdd.collect())
print(rdd2)
20
10
30
40
50
None


12. saveasTextFile()

rdd &＃61; sc.parallelize([1, 3, 2, 4, 7, 9, 6], 1)# 分区可以控制生成的文件数
# rdd.saveAsTextFile("../data/output/out1") 保存写相对路径也不行 还是要绝对路径
# 本地
rdd.saveAsTextFile(f"file:///{ROOT}/data/output/out1")
# hdfs
rdd.saveTextFile("hdfs://hadoop102:8080/路径")


foreach; saveAsTextFile 都是由分区&＃xff08;Excutor)直接执行的,跳过Driver

其余的 Action 算子都会将结果发送至Driver