零基础入门大数据之spark中rdd部分算子详解

àì夳堔傛蜴生んèń 2023-07-24 14:52 9阅读 0赞

零基础入门大数据之spark中rdd部分算子详解先前文章介绍过一些spark相关知识，本文继续补充一些基础算子，主要包括：

1. parallelize

2. aggregate

3. cache

4. cartesian

5. distinct

6. filter

7. keyBy

我们知道，spark中一个重要的数据结构是rdd，这是一种并行集合的数据格式，大多数操作都是围绕着rdd来的，rdd里面拥有众多的方法可以调用从而实现各种各样的功能，那么通常情况下我们读入的数据源并非rdd格式的，如何转换为rdd呢？

一个基本的方法是初始化，或者格式化操作函数parallelize。

*   parallelize

比如一个数组Array(1,2,3,4,5),经过parallelize后就变成了rdd格式的数组。

scala> val d = Array(1,2,3,4,5)
    d: Array[Int] = Array(1, 2, 3, 4, 5)
    
    scala> val rdd_d = sc.parallelize(d)
    rdd_d: org.apache.spark.rdd.RDD[Int] = ParallelCollectionRDD[2] at parallelize at <console>:27

rdd的初始化parallelize函数可以接受两个参数，上面省略了第二个默认参数，这个参数是分片个数slices，表示数据集切分的份数。一份数据存在某台机器上的时候，可以指定切分的份数，可以想象，切分的越多，处理起来因为是并行的，速度越快，当然这是要占资源的，当数据小的时候完全没必要。典型地，你可以在集群的每个CPU上分布2-4个slices. 一般来说，Spark会尝试根据集群的状况，来自动设定slices的数目。也可以通过传递给parallelize的第二个参数来进行手动设置，例如：sc.parallelize(data, 10))。

*  aggregate

再来看一个稍复杂的函数：aggregate。这是一个比较底层的使用也很广泛的函数，字面意思就是聚合的意思。不过它有两层聚集：第一层，对分片的数据进行聚集，这里的片就是上面的slices，第二层，对聚集的结果再进行聚集。看一下函数原型：

def aggregate[U: ClassTag](zeroValue: U)(seqOp: (U, T) => 
    U, combOp: (U, U) => U): U)

这个函数需要初始化一个值zeroValue。第一层聚合seqOp，第二层聚合combOp。

看一个例子：

scala> val d = Array(1,2,3,4,5,6)
    scala> val rdd_d = sc.parallelize(d,2)
    scala> val res = rdd_d.aggregate(0)(math.max(_,_),_+_)
    
    res: Int = 9

解释一下：

1）首先将一组数组分成两片存储起来：parallelize(d,2)；默认是平均分，也就是1，2，3一片，4，5，6一片

2）对每一片执行取最大值操作，因为是聚合函数，所以输入是两个，输出是一个的函数，math.max(\_,\_)就是这种，下划线代表所有元素执行。可以看到这一步过后每一片的结果就是3，6。

3）之后对3，6执行相加的操作得到9.

一般来说这两个函数设置成一样的，比如，就是取最大值，那么第二个函数也可以变成math.max(\_,\_)

*  cache方法

cache是将RDD的结果暂时存放在内存里面，方便后面需要用到这个rdd的时候不用再计算。

我们知道，spark里面包含两种运算算子，一种是转换算子，一种是行为算子，转换算子再spark里面是惰性计算的，什么意思呢？就是你写了代码，但是实际上程序运行到这一步并没有实际发生计算，只有碰到了行为算子才算正儿八经的计算。转换算子，就好比画了计算流程图一样，只是单纯的框架而已。这让我想起了tensorflow里面的一些运算也是这样，果然都是一家的程序员，思路都完美继承。

说远了，这和cache算子有什么关系呢？举个例子，假设数据a经过三层map变成了d，也就是a->b->c->d，a到d之间都是转换操作，这个时候如果某个计算需要用到d，那么就会把a到d的过程重新走一遍，因为d在一次计算以后不会保存在内存里面的，这就导致了一个严重的问题就是需要重复计算很多东西。如果d被程序在不同地方多次调用的话将带来性能的下降。这个时候有没有办法把d第一次被计算的结果保存起来呢，有，这就是cache方法，起到缓存的作用。

*  笛卡尔操作 cartesian

什么是笛卡尔操作？就是两个集合中的元素分别两两排列组合，举个例子：

val s1 = sc.parallelize(Array(1,2,3,4,5))
    val s2 = sc.parallelize(Array("a","b","c"))
    val res = s1.cartesian(s2)
    res.collect()
    
    >>>  Array[(Int, String)] = Array((1,a), (1,b), (1,c), (2,a), (2,b), 
    (2,c), (3,a), (3,b), (3,c), (4,a), (4,b), (4,c), (5,a), (5,b), (5,c))

看例子非常容易理解。笛卡尔操作在某些时候还是很有效的。

*  去重方法：distinct

顾名思义，去掉rdd中重叠的元素。实际中我经常也会用到，比如想把几个rdd进行合并起来，可以用union方法，但是呢里面会有重复的，这个时候再接这个函数即可。如下：

val s1 = sc.parallelize(Array(1,2,3,4,5))
    val s2 = sc.parallelize(Array(4,5,6,7,8))
    val res = s1.union(s2).distinct()
    res.collect()
       
    >>> Array[Int] = Array(1, 2, 3, 4, 5, 6, 7, 8)

*  过滤方法：filter

这个方法可以说非常的重要了，灵活运用可以实现非常多的功能。filter方法就是过滤掉rdd中满足一定条件的rdd，filter函数里面可以定义各种过滤函数。比如下面：

val s1 = sc.parallelize(Array(1,2,3,4,5))
    val res = s1.filter(x => x>3)
    res.collect()
    
    >>> Array[Int] = Array(4, 5)

*   生成键值对的keyBy方法

keyBy方法是为rdd中的每个数据额外增加一个key构成（key，value）的数据对，进而这种结构的数据可以使用（key，value）专门的一些聚合函数，这些函数在以前的文章中记录过。keyBy的方法也比较简单，举个例子就明白了：如果你对大数据开发感兴趣，想系统学习大数据的话，可以加入大数据技术学习交流扣扣君羊：**522189307**

val s1 = sc.parallelize(Array(1,2,3,4,5))
    val res = s1.keyBy(x => x*x)
    res.collect()
    
    >>> Array[(Int, Int)] = Array((1,1), (4,2), (9,3), (16,4), (25,5))

这里将rdd中每个数的平方值当作自己的key，生成的结果可以看到。所以keyBy只是生成对应元素key。