[spark]transformation算子

news/2024/5/4 10:57:01/文章来源:https://blog.csdn.net/qq_35896718/article/details/127525635

1.sample算子

1）说明

sample算子：对rdd中的数据进行抽样。一个非常重要的作用就是开看rdd中数据的分布，进行各种调优与优化。

数据倾斜：数据分布的不均匀，shuffle会将相同key的数据汇总到一台机器上，就会导致某些task执行的特别慢。

找到哪些发生数据倾斜的key：sample算子+reduceByKey就可以知道哪一个key出现次数最多，出现次数最多的key往往就是发生数据倾斜的key，找到这个key后就可以进行数据倾斜优化。

如果用reduceByKey显然能统计出key的次数，但是该方法本很还是会造成数据倾斜，所以不采用这种方式。

2）用法

参数：

withReplacement：是否可替换，确定抽样方法。true为可放回抽同样，false为不可放回抽样。fraction(小数):确定抽样比例，范围(0,1)

seed：种子数

2.map、mapPartition、mapPartitionWithIndex

1)map(f:T=>U)

对rdd中每个元素调用一次f函数，效率低。该函数是有一个参数，有一个返回值的函数。

输出：

说明：map和foreach都是窄依赖，所以task是并行执行的，分区是按照Range进行分区的，并且有的task执行的快，有的task执行的慢，所以输出顺序不确定。

2)mapPartitions(f:Iterator[T]=>Iterator[U])

该函数是对该rdd被分成的每一个partition调用一次f函数，效率比map高。

迭代器：可遍历的东西就是迭代器，比如集合。

输出:3 5 7 2 8 6 4

说明: partition=>parititon.map(_*7)是参数和返回值都是迭代器的函数

3)mapPartitionsWithIndex(f:(Int,Iterator[T])=>Iterator[U])

该函数也是对该rdd被分成的每一个partition调用一次f函数，该函数的参数是：(index,迭代器)即分区号和迭代器，返回值是迭代器。该算子和mapPartition的区别是：多了一个index，该index代表的就是当前被操作的paritition的编号，一个rdd被分为多个partition，每个partition都有一个编号，编号从0开始，假如有N个分区，编号的取值范围就是[0,N-1]，通过该index可以清楚的看到每一个分区中数据的分布。

输出

说明:

mapPartitionsWithIndex算子和foreach算子都是窄依赖，即task是并行执行的。

分区是按照Range进行分区的，有的task执行的快有的task执行的慢，所以每次输出顺序不一样。

3.union(窄依赖)

union(other:RDD[T]):RDD[T]

hive中:union去重且排序，union all不去重不排序

spark的union相当于hive的union all

union只是将两个rdd的分区进行了拼接，它是一个窄依赖

4.distinct(宽依赖)

作用:将所有分区内的元素收集到一块进行去重，再将去重后的数据进行分区，参数指的是分多少个分区。

输出

[0]List(1, 3, 7, 9, 5)

5.intersection(宽依赖)

6.groupByKey(宽依赖)、groupBy(宽依赖)

1)groupByKey(numPartitions:Int):RDD[K,Iterable[V]]

作用：按照key将数据进行分组，groupByKey操作的RDD的数据类型是k-v类型的。

参数numPartitions：指结果RDD的分区个数，也可以不写参数，默认是和父RDD一样的分区数，返回值是一个元组，第一个值是key的值，第二个值是该key进行分组的value组成的迭代器。

reduceByKey是有本地预聚合操作的，而这个groupByKey是没有本地预聚合操作的。所以不能groupByKey就不要用它，可以用reduceByKey或aggregateByKey替代。

2)groupBy(f:T=>K):RDD[(K,Iterable[T])]

内部调用的是groupByKey,最好用reduceByKey或aggregateByKey代替。

和groupByKey不同的是它操作的不是k-v类型,它的参数是一个函数f，这个函数的返回值K会作为分组的key，value是f函数中的参数T，groupBy按照这个key进行分组。groupBy的返回值和groupByKey一毛一样。

输出:

河北:List(1,小红)

北京:List(2,小明)

黑龙江:List(3,小灰, 4,小微)

输出:
黑龙江:List(3,小灰,黑龙江, 4,小微,黑龙江)

北京:List(2,小明,北京)

河北:List(1,小红,河北)

7.join(两个rdd的分区相同时是窄依赖，不同时是宽依赖)

join操作的必须是k-v类型的RDD

1）across join交叉连接(笛卡尔积)

2）内连接

参数是一个k-v格式的RDD，返回值也是一个元组，元组的第一个值是key，第二个值是两个集合的value组成的元组。

A join B on A.id=B.id

join也可以写成inner join。获取A和B中能够关联上的数据。得到的结果一定是一定的。

3）外连接

外连接的输出数据中一张表中的数据是不确定的，可能有，可能没有。

可能有可能没有的数据类型为Option，Option只有两个值Some(有)，None(没有)

a)leftOuterJoin左外连接

左表数据原样输出，右表关联上的显示，关联不上的显示null。

b)rightOuterJoin右外连接

右外连接，右表数据原样输出，左边关联上的显示，关联不上的显示为null。

c)fullOuterJoin全外连接

分别以两张表为基准表，关联上的显示，关联不上的显示为null。

4）半连接(spark不支持半连接，但Hive支持半连接)

5）例子

输出:

8.cogroup

1）

rdd1.groupByKey:将rdd1按照它的key进行分组，返回key，value组成的元组。其中value是个迭代器，是相同key对应的值组成的迭代器。

2）

cogroup这个单词的意思：在两个或多个关系中的分组

作用：就是对两个K-V键值对分别执行groupByKey操作，cogroup产生的结果都在一个分区中。

rdd1.cogroup(rdd2):将rdd1和rdd2按照他们相同的key进行分组，返回key,rdd1的value,rdd2的value组成的元组，在这个元组中._1是key，._2是(value1,value2)，其中rdd1的value，rdd2的value都是迭代器，是相同key对应的值组成的迭代器。