Spark学习摘记 —— Spark转化操作API归纳

Posted 2021-03-04 枯落

本文参考

在阅读了《Spark快速大数据分析》动物书后，大概了解到了spark常用的api，不过书中并没有给予所有api详细的案例，而且现在spark的最新版本已经上升到了2.4.5，动物书中的spark版本还停留在1.2.0版本，所以就有了这篇文章，在最新的2.4.5版本下测试常用的api

由于spark的惰性计算特性，RDD只有在第一次行动操作中被用到时才会真正进行计算，因此我打算将文章内容分为"转化操作API"和"行动操作API"两部分，同时因为pair RDD（RDD中的元素是键值对）的部分api较为特殊，所以我打算单独再写一篇文章

本文仅介绍转化操作API，前5个api —— map()、flatMap()、filter()、distinct()、sample()是针对一个RDD的转化操作，后续的api —— union()、intersection()、subtract()、cartesion()是针对两个RDD的转化操作

环境

idea + spark 2.4.5 + scala 2.11.12

RDD均通过SparkContext的parallelize()函数创建

map()函数

目的：

将函数应用于RDD中的每个元素，将返回值构成新的RDD

转化前后的RDD的元素类型可以不同（比如经典的WordCount示例中转化为了键值对元素）

代码：

val testList = List(1, 2, 3, 3)

val testRdd = sc.parallelize(testList)

testRdd.map(ele => ele * ele).foreach(ele => print(s"$ele "))
																								

输出：

1 4 9 9

更高效的操作：

每个RDD被分为多个分区，这些分区在集群的不同节点上运行，可以使用mapPartitions()函数，将转化操作作用于每个分区的元素上，这种方法还可以为每个分区创建一个JDBC连接，而不是为每一个元素创建一个连接（此处不做示例）

mapPartitions()函数有两个参数，第一个参数接收一个函数，和map()函数相同，第二个参数为preservesPartitioning，默认值为false，仅当我们对pair RDD进行转化操作，并且没有修改键时设置为true

val testList = List(1, 2, 3, 3)

val testRdd = sc.parallelize(testList)

testRdd.mapPartitions(partition =>

					partition.map(

					ele => {

					ele * ele

					}

)).foreach(ele => print(s"$ele "))

?

flatMap()函数

目的：

将函数应用于RDD中的每个元素，将返回的迭代器的所有内容构成新的RDD，我们也常常说成是"压扁"

"压扁"这个词可能听上去不大好理解，我们提供给flatMap()的函数分别应用到RDD的那个元素上，不过返回的不是一个元素，而是一个返回值序列的迭代器，但输出的RDD不是由迭代器组成，得到的是一个包含各个迭代器可以访问的所有元素的RDD

转化前后的RDD的元素类型不变

代码：

val testList = List(1, 2, 3, 3)

val testRdd = sc.parallelize(testList)

testRdd.flatMap(ele => {

					ele.to(5)

}).foreach(ele => print(s"$ele "))
																	

我们也可以手动返回迭代器，这段代码也类似于

val testList = List(Range(1, 6), Range(2, 6), Range(3, 6), Range(3, 6))

val testRdd = sc.parallelize(testList)

testRdd.flatMap(_.iterator).foreach(ele => print(s"$ele "))
																								

输出：

1 2 3 4 5 2 3 4 5 3 4 5 3 4 5

?

filter()函数

目的：

返回一个由传给filter()函数的元素组成的RDD，当函数返回值为true时，保留该元素，可以理解为 "被过滤"出来

代码：

val testList = List(1, 2, 3, 3)

val testRdd = sc.parallelize(testList)

testRdd.filter(ele => ele > 2).foreach(ele => print(s"$ele "))
																									

输出：

3 3

更高效的操作：

通过过滤操作后，RDD中的元素减少，可以在filter()操作后执行coalesce()函数进行分区合并，第一个参数指定分区数，当指定的分区数大于当前RDD的分区数时不会进行合并，当前分区数不变（除非指定第二参数shuffle为true，默认为false），当指定的分区数小于当前的RDD的分区数时会进行合并，并且不会进行shuffle（尽量不要指定极端的情况，如指定合并后的分区数为1）

val testList = List(1, 2, 3, 3)

val testRdd = sc.parallelize(testList)

testRdd.filter(ele => ele > 2).coalesce(5).foreach(ele => print(s"$ele "))

?

distinct()函数

目的：

去重，因为会进行shuffle，所以不推荐此操作

代码：

val testList = List(1, 2, 3, 3)

val testRdd = sc.parallelize(testList)

testRdd.distinct().foreach(ele => print(s"$ele "))
																				

输出：

1 2 3

?

sample()函数

目的：

对RDD进行采样

代码：

val testList = List(0, 1, 2, 3, 4, 5, 6, 7, 8, 9)

val testRdd = sc.parallelize(testList)

testRdd.sample(false, 0.9).foreach(ele => print(s"$ele "))
																								

第一个参数withReplacement指定false时，第二个参数fraction必须为 [ 0 , 1 ] 之间，表示每个元素被选中的可能性

按照该示例，也有人将该函数理解为，从所有元素中抽取90%返回，但是在源码中已经我们可以看到"without replacement: probability that each element is chosen; fraction must be [0, 1]"，并且也指明"This is NOT guaranteed to provide exactly the fraction of the count of the given [[RDD]]"，因此这种理解方式我认为是错误的

输出：

0 1 2 3 5 6 8 9（不一定）

疑点：

当第一个参数withReplacement指定true时，第二个参数fraction并不要求一定小于1，源码中注释为"with replacement: expected number of times each element is chosen; fraction must be greater than or equal to 0"

val testList = List(0, 1, 2, 3, 4, 5, 6, 7, 8, 9)

val testRdd = sc.parallelize(testList)

testRdd.sample(true, 3).foreach(ele => print(s"$ele "))
																								

输出：0 0 1 1 1 1 2 2 2 3 3 3 3 4 4 4 4 4 5 5 5 5 5 5 6 6 7 7 8 9 9

目前不大理解是如何在采样的，希望各位看官大大能在评论区发表看法哈

===============暂时更新到这儿哈==============

?

?