Spark开发学习之RDD编程

什么是RDD

RDD(Resilient Distributed Dataset)是分布式数据集，是Spark设计里最为核心的概念。在RDD出来之前，所有的分布式批处理计算系统都是从存储中读取数据到计算完成后将结果写入存储的模型，这种计算模型在处理数据集迭代运算时效率不高，为了解决这一问题，RDD应运而生。

在这里插入图片描述

如上图，RDD是Spark core层最重要的概念，其他的Spark SQL , Spark streaming 等都是基于RDD去计算处理的。

spark将数据抽象成RDD模型，spark对相关问题的解决都是通过对RDD进行多次操作来实现的，所以spark的核心即是RDD编程。

RDD编程知识图谱

在这里插入图片描述

RDD创建

RDD创建通常有两种方式，基于加载文件创建和使用集合数据类型创建。

加载文件时创建

RDD编程中一个重要对象是SparkContext。在启动spark-shell时会自动创建一个sc的SparkContext对象，可以在shell里直接使用。如下：

sc.textFile()方法加载文件，生成内存中的RDD,加载文件支持本地文件系统文件，hdfs文件，网络文件，如下：
```
scala> val rdd = sc.textFile("/home/bigData/softnew/spark/README.md")
rdd: org.apache.spark.rdd.RDD[String] = /home/bigData/softnew/spark/README.md MapPartitionsRDD[3] at textFile at <console>:24
```
使用集合数据类型创建

定义的集合类型数据需要使用parallelize()函数进行序列化为RDD对象，如下数组元素被序列化为含四个元素的RDD对象：

在这里插入图片描述

RDD 基本操作

RDD 基本操作分为两种操作，转换（Transformation）和动作（Action）。
所有的业务都是使用这两种操作去实现的，业务流程如下：

在这里插入图片描述

但RDD计算是惰性的，即转换期间并不会发生计算，而是保存转换记录，到动作发生时按照记录进行计算。这种设计保证了spark计算的高效。

转换操作API常见的有以下几种：

map(func) 将每个元素传递func函数中，并将结果返回一个新的数据集
filter(func) 筛选出满足函数func的元素，并返回一个新的数据集
flatMap() 基于map结果，将每个RDD里的列表元素拆开形成多个RDD
groupByKey() 作用于(K,V)数据集，返回新的（K,Iterable）形式数据集（按key聚合）
reduceByKey(func)

map

map(func)是一个高阶函数，里面可以带lamb表达式函数。对rdd中每一个元素进行加5操作，图示如下：

在这里插入图片描述

filter

filter(func)是一个高阶函数，里面可以带lamb表达式函数。筛选出包含best字符的RDD元素，图示如下：
在这里插入图片描述

flatMap()

flatMap(func)是在map的基础上加了一个flat,即把每个RDD元素中的数组元素拆开成多个RDD元素，图示如下：
在这里插入图片描述

图中1、2步的操作即是flatMap()函数的业务流程。该示例输入一个文本，通过flatmap函数将文本转成一个个单次RDD,便于后续词频统计。

groupByKey

groupByKey()是低阶函数,对(K,V)集合数据按K进行分组，返回值列表,如下所示：
在这里插入图片描述

reduceByKey

reduceByKey(func)是高阶函数，它是在groupByKey()基础上，利用func定义函数对值列表进行计算，并返回结果，如下所示：
在这里插入图片描述

其中，红色实线的reduceByKey((a,b)=>a+b),分为两步，如蓝色虚线所示，groupByKey()和(a,b)=>a+b。值列表为两个元素的很好理解，直接按照a,b传入参数，相加返回即可。那三个参数是如何进行相加的嘞？

比如 (is,(1,1,1)) 调用(a,b)=>a+b 流程如下：

第一个1和第二个1分别传给 a、b，求和结果作为a 和第三个1 分别再次传给a、b,求和即为最终结果3,其它复杂值列表类似处理。

动作操作API常见的有以下几种：

count() 统计当前RDD数据集中元素个数
collect() 将分布在不同电脑上的集合元素都收集回来，在客户端显示，返回数据集
first() 返回数据集第一个元素
take(n) 以数组形式返回数据集前n个元素
reduce(func) 通过func函数来聚合数据集
foreach(func) 将数据集中每个元素传递单func函数执行，比如传递println()函数，依次打印数据集元素

动作函数比较简单，下面看一个示例（在spark-shell下完成）：

scala> val array = Array(1,2,3,4,5)
array: Array[Int] = Array(1, 2, 3, 4, 5)

scala> val rdd = sc.parallelize(array)
rdd: org.apache.spark.rdd.RDD[Int] = ParallelCollectionRDD[4] at parallelize at <console>:26

scala> rdd.first()
res4: Int = 1

scala> rdd.take(4)
res5: Array[Int] = Array(1, 2, 3, 4)

scala> rdd.reduce((a,b)=>a+b)
res6: Int = 15

scala> rdd.foreach(em=>println(em))
5
3
2
1
4

RDD 持久化

由于RDD计算是有惰性机制的，即所有转换操作只有在动作操作时才触发执行，在动作操作之前只会记录所有转换操作。所以整个过程中是没有中间变量一说的，若想保留某个计算中间结果，方便后面再次使用，需要使用RDD的持久化API。这样也可以提高程序执行效率，但持久化不能滥用，只有在确认会多次使用的时候才去做持久化。

persist(param) 将RDD标记为持久化

其中，param可以支持如下两个参数，

MEMORY_ONLY, 放到内存中，内存不足时会覆盖旧的；

MEMORY_AND_DISK 放到内存中，内存不足会放到磁盘中，不会覆盖旧的数据。

persist(MEMORY_ONLY) 可以简写为 cache()
unpersist() 将RDD数据集移出缓存

ps:以上持久化相关操作都是等到遇到动作操作时才真正去执行持久化。