RDD：常用Transform算子和Action算整理一、Transform算子二、Action算子

提示：文章写完后，目录可以自动生成，如何生成可参考右边的帮助文档

一、Transform算子

根据数据处理方式，可以分为value类型，双value类型，和 Key-Value类型

1）value类型

map

和python差不多，map转换就是对每一个元素进行一个映射转换，可以是数值的转换，也可以是类型的转换

#pyspark
rdd = sc.parallelize([1,2,3]) 
rdd_map = rdd.map(lambda num:num*2)
print(rdd_map.collect())
#2,4,6

mapPartition

以分区为单位处理数据
存在问题：整个分区数据会被加载到内存中引用，由于存在对象的引用，已处理完的部分数据不会被释放掉，数据量大，内存小的会导致内存溢出

rdd = sc.parallelize([1,2,3])
rdd_par = rdd.mapPartitions(lambda num:num*2)

map和mapPartiton的区别
1.数据处理方面：map来一个处理了一个，mapPartition以分区为单位进行批处理
2.功能方面：map主要目的是数据的转换与改变，不会改变数据量；mapPartition传递一个迭代器，返回一个迭代器，没有要求数据的个数保持不变
3.性能方面：map类似于串行操作，性能比较低，mapPartition类似于批处理，性能较高，但可能会造成数据溢出

mapPartitionWithIndex

根据分区号取分区数据/查数据所在的分区

flatMap

扁平化操作 list–>int

    rdd = sc.parallelize([[1,2],[3,4],3,[4,5,6]],numSlices=2)
    print(rdd.collect())
    #[[1,2],[3,4],3,[4,5,6]]
    rdd_flatmap = rdd.flatMap(lambda x:x if isinstance(x,Iterable) else (x,))
    print(rdd_flatmap.collect())
    #[1, 2, 3, 4, 3, 4, 5, 6]
    rdd_per = rdd_flatmap.mapPartitionsWithIndex(lambda par_num,v:(par_num,list(v)) if par_num==1 else iter(''))
    print(rdd_per.collect())
    #[1, [3, 4, 5, 6]]

glom

int–>list
eg:计算分区内最大值求和（分区内取最大值，分区间求和）

groupBy、sortBy

groupBy：根据Key聚合数据，分区不变，数据打乱重新组合
sortBy：用于排序数据。在排序之前，可以将数据通过 f 函数进行处理，之后按照 f 函数处理
的结果进行排序，默认为升序排列。排序后新产生的 RDD 的分区数与原 RDD 的分区数一
致。中间存在 shuffle 的过程

filter

将数据根据指定的规则进行筛选过滤，符合规则的数据保留，不符合规则的数据丢弃。
当数据进行筛选过滤后，分区不变，但是分区内的数据可能不均衡，生产环境下，可能会出
现数据倾斜。

distinct

去重

#去重原理
map(lambda x: (x, None)).reduceByKey(lambda x, _: x, numPartitions).map(lambda x: x[0])

coalesce，repartition

coalesce：根据数据量缩减/扩大分区，用于大数据集过滤后，提高小数据集的执行效率

#numPartitions：分区数
#shuffle：是否将数据打乱重新组合，使新分区数据均衡，设为True时可以用于扩大分区
coalesce(numPartitions, shuffle=False)

当 spark 程序中，存在过多的小任务的时候，可以通过 coalesce 方法，收缩合并分区，减少
分区的个数，减小任务调度成本
repartition：扩大分区，底层代码调用的是coalesce(self, numPartitions, shuffle=True)

2）双value类型

intersection、union、subtract、zip

交集、并集、差集
拉链：相同位置的数据放在一起，输出为tuple类型。注：两个数据源分区数量一致、分区中数据的数量也保持一致。

3）Key-Value类型

partitionBy

将数据按照指定 Partitioner 重新进行分区。Spark 默认的分区器是 HashPartitioner

reduceBykey、groupBykey

reduceBykey:按key现在分区内进行聚合，之后在进行分区间聚合；因此分区内、分区间计算规则要一致

groupBykey：按key进行分组，聚合发生在分区间

两个的区别
从Shuffle角度：
两者都存在shuffle，reduceBykey在shuffle前，将key值相同的value进行了一次聚合，减少了shuffle阶段落盘的数据量；groupBykey只是进行分组，不存在数据量减少的问题
从功能的角度：yBKey 只能分组，不能聚
合

aggregateBykey

将数据根据不同的规则进行分区内计算和分区间计算，用户返回值的类型与输入一致。

foldByKey

当分区内计算规则和分区间计算规则相同时，aggregateByKey 就可以简化为 foldByKey

combinBykey

最通用的对 key-value 型 rdd 进行聚集操作的聚集函数（aggregation function）。类似于
aggregate()，combineByKey()允许用户返回值的类型与输入不一致。

reduceByKey、foldByKey、aggregateByKey、combineByKey 的区别？
reduceByKey: 相同 key 的第一个数据不进行任何计算，分区内和分区间计算规则相同
FoldByKey: 相同 key 的第一个数据和初始值进行分区内计算，分区内和分区间计算规则相
同
AggregateByKey：相同 key 的第一个数据和初始值进行分区内计算，分区内和分区间计算规
则可以不相同
CombineByKey:当计算时，发现数据结构不满足要求时，可以让第一个数据转换结构。分区
内和分区间计算规则不相同。

join

在类型为(K,V)和(K,W)的 RDD 上调用，返回一个相同 key 对应的所有元素连接在一起的
(K,(V,W))的 RDD

两个不同数据源，相同key的value会连接在一起，形成tuple
如果两个数据源中key没有匹配上，那么数据不会出现在结果中
如果两个数据源中key有多个相同的，会一次匹配，出现笛卡尔积，导致性能降低

leftOuterJoin

类似于表的左连接

cogroup：connect+group

在类型为(K,V)和(K,W)的 RDD 上调用，返回一个(K,(Iterable,Iterable))类型的 RDD
首先如果一个RDD中有相同的key，现在同一个Rdd中分组，之后再RDD间相同的key链接

二、Action算子

collect

在驱动程序中，以数组 Array 的形式返回数据集的所有元素

first、take(n)、takeOrdered

取第一个元素
取前n个元素
返回该 RDD 排序后的前 n 个元素组成的数组

reduce

聚集 RDD 中的所有元素，先聚合分区内数据，再聚合分区间数据

aggregate

分区的数据通过初始值和分区内的数据进行聚合，然后再和初始值进行分区间的数据聚合

fold

折叠操作，aggregate 的简化版操作

saveAsTextFile

保存rdd成text文件到本地

countByKey

统计每种 key 的个数

foreach

分布式遍历 RDD 中的每一个元素，调用指定函数
RDD算子API

最后

以上就是强健信封为你收集整理的RDD：常用Transform算子和Action算整理一、Transform算子二、Action算子的全部内容，希望文章能够帮你解决RDD：常用Transform算子和Action算整理一、Transform算子二、Action算子所遇到的程序开发问题。

如果觉得靠谱客网站的内容还不错，欢迎将靠谱客网站推荐给程序员好友。