Spark学习—RDD编程

╰半橙微兮° 2022-06-16 07:18 204阅读 0赞

RDD:弹性分布式数据集(ResilientDistributed Dataset),是Spark对数据的核心抽象。RDD其实是分布式的元素集合。当Spark对数据操作和转换时，会自动将RDD中的数据分发到集群，并将操作并行化执行。

Spark中的RDD是一个不可变的分布式对象集合。每个RDD都倍分为多个分区，这些分区运行在集群中的不同节点。RDD可以包含Python、Java、Scala中任意类型的对象，甚至可以包含用户自定义对象，本文主要通过Java实现相关示例。

Spark程序或shell会话工作流程

1.    从外部数据创建出输入RDD;

2.    使用诸如filter()等这样的转化操作对RDD进行转化，以定义新的RDD;

3.    告诉Spark对需要被重用的中间结果RDD执行persist()操作；

4.    使用诸如first()等这样的行动操作来触发一次并行计算，Spark会对计算进行优化后再执行。

# 一.  创建RDD #

Spark提供了两种创建RDD方式：

1.    读取外部数据集，如文件，hive数据库等；

2.    在驱动器程序中对一个集合进行并行化，如list,set等。

方法1是常用方法，其从外部存储中读取数据来创建RDD,如读取文件

![1240][]

方法1创建RDD

方法2其实使用较少，毕竟它需要把整个数据集先放在一台机器的内存中。实现也简单，就是把程序中一个已有集合传给SparkContext的parallelize()方法。二.RDD操作

![1240 1][]

方法2创建RDD

# 二.  RDD操作 #

### 1.    RDD支持两种操作 ###

(1)  转化操作，RDD的转化操作是返回一个新的RDD的操作，比如map()和filter。

(2)  行动操作，RDD的行动操作则是向驱动器程序返回结果或把结果写入外部系统的操作，会触发实际的计算，比如count()和first()。

惰性求值：RDD的转化操作是惰性求值的，即在被调用行动操作之前Spark不会开始计算，相反，Spark会在内部记录下索要求执行的操作的相关信息。例如，当我们调用jsc.textFile()时，数据并没有读取进来，而是在必要时才会读取。Spark使用惰性求值，就可以把一些操作合并到一起来减少计算数据的步骤。

### 2.    RDD的基本转化操作 ###

对一个数据为\{1,2,3,3\}的RDD进行基本的RDD转化操作

对数据分别为\{1, 2,3\}和\{3, 4, 5\}的RDD进行针对两个RDD的转化操作

### 3.    RDD的基本执行操作 ###

<table> 
 <tbody> 
 <tr> 
 <td> 函数名 </td> 
 <td> 目的 </td> 
 <td> 示例 </td> 
 <td> 结果 </td> 
 </tr> 
 <tr> 
 <td> collect() </td> 
 <td> 返回RDD 中的所有元素 </td> 
 <td> rdd.collect() </td> 
 <td> {1, 2, 3, 3} </td> 
 </tr> 
 <tr> 
 <td> count() </td> 
 <td> RDD 中的元素个数 </td> 
 <td> rdd.count() </td> 
 <td> 4 </td> 
 </tr> 
 <tr> 
 <td> countByValue() </td> 
 <td> 各元素在RDD 中出现的次数 </td> 
 <td> rdd.countByValue() </td> 
 <td> {(1, 1),(2, 1),(3, 2)} </td> 
 </tr> 
 <tr> 
 <td> take(num) </td> 
 <td> 从RDD 中返回num 个元素 </td> 
 <td> rdd.take(2) </td> 
 <td> {1, 2} </td> 
 </tr> 
 <tr> 
 <td> top(num) </td> 
 <td> 从RDD 中返回最前面的num个元素 </td> 
 <td> rdd.top(2) </td> 
 <td> {3, 3} </td> 
 </tr> 
 <tr> 
 <td> takeOrdered(num) (ordering) </td> 
 <td> 从RDD 中按照提供的顺序返回最前面的num 个元素 </td> 
 <td> rdd.takeOrdered(2)(myOrdering) </td> 
 <td> {3, 3} </td> 
 </tr> 
 <tr> 
 <td> takeSample(withReplacement, &nbsp;num, [seed]) </td> 
 <td> 从RDD 中返回任意一些元素 </td> 
 <td> rdd.takeSample(false, 1) </td> 
 <td> 非确定的 </td> 
 </tr> 
 <tr> 
 <td> reduce(func) </td> 
 <td> 并行整合RDD 中所有数据（例如sum） </td> 
 <td> rdd.reduce((x, y) =&gt; x + y) </td> 
 <td> 9 </td> 
 </tr> 
 <tr> 
 <td> fold(zero)(func) </td> 
 <td> 和reduce() 一样， 但是需要提供初始值 注意:不重复元素加初始值，重复元素只加一个 </td> 
 <td> rdd.fold(0)((x, y) =&gt; x + y) </td> 
 <td> 9 </td> 
 </tr> 
 <tr> 
 <td> aggregate(zeroValue) (seqOp, combOp) </td> 
 <td> 和reduce() 相似， 但是通常返回不同类型的函数 注意:不重复元素加初始值，重复元素只加一个 </td> 
 <td> rdd.aggregate((0, 0))((x, y) =&gt; (x._1 + y, x._2 + 1),(x, y) =&gt; (x._1 + y._1, x._2 + y._2)) </td> 
 <td> (9,4) </td> 
 </tr> 
 <tr> 
 <td> foreach(func) 对RDD </td> 
 <td> 中的每个元素使用给定的函数 </td> 
 <td> rdd.foreach(func) </td> 
 <td> 无 </td> 
 </tr> 
 </tbody> 
</table>

对一个数据为\{1, 2,3, 3\}的RDD进行基本的RDD行动操作

### 4.    标准Java函数接口 ###

在Java中，函数需要作为实现了Spark的org.apache,spark.api.java.function包中的任一函数接口的对象传递。

<table> 
 <tbody> 
 <tr> 
 <td> 函数名 </td> 
 <td> 实现的方法 </td> 
 <td> 用途 </td> 
 </tr> 
 <tr> 
 <td> Function&lt;T, R&gt; </td> 
 <td> R call(T) </td> 
 <td> 接收一个输入值并返回一个输出值，用于类似map() 和filter() 等操作中 </td> 
 </tr> 
 <tr> 
 <td> Function2&lt;T1, T2, R&gt; </td> 
 <td> R call(T1, T2) </td> 
 <td> 接收两个输入值并返回一个输出值，用于类似aggregate()和fold() 等操作中 </td> 
 </tr> 
 <tr> 
 <td> FlatMapFunction&lt;T, R&gt; </td> 
 <td> Iterable&lt;R&gt; call(T) </td> 
 <td> 接收一个输入值并返回任意个输出，用于类似flatMap()这样的操作中 </td> 
 </tr> 
 </tbody> 
</table>

标准Java函数接口

### 5.    Java中针对专门类型的函数接口 ###

<table> 
 <tbody> 
 <tr> 
 <td> 函数名 </td> 
 <td> 等价函数 </td> 
 <td> 用途 </td> 
 </tr> 
 <tr> 
 <td> DoubleFlatMapFunction&lt;T&gt; </td> 
 <td> Function&lt;T, Iterable&lt;Double&gt;&gt; </td> 
 <td> 用于flatMapToDouble，以生成DoubleRDD </td> 
 </tr> 
 <tr> 
 <td> DoubleFunction&lt;T&gt; </td> 
 <td> Function&lt;T, Double&gt; </td> 
 <td> 用于mapToDouble，以生成DoubleRDD </td> 
 </tr> 
 <tr> 
 <td> PairFlatMapFunction&lt;T, K, V&gt; </td> 
 <td> Function&lt;T, Iterable&lt;Tuple2&lt;K, V&gt;&gt;&gt; </td> 
 <td> 用于flatMapToPair，以生成PairRDD&lt;K, V&gt; </td> 
 </tr> 
 <tr> 
 <td> PairFunction&lt;T, K, V&gt; </td> 
 <td> Function&lt;T, Tuple2&lt;K, V&gt;&gt; </td> 
 <td> 用于mapToPair， 以生成PairRDD&lt;K, V&gt; </td> 
 </tr> 
 </tbody> 
</table>

Java中针对专门类型的函数接口

# 三.  示例 #

本节将通过示例的方式验证第二节中相关的转化操作和行动操作。

![1240 2][]

转化和执行计算结果

代码地址：[http://download.csdn.net/detail/a123demi/9837375][http_download.csdn.net_detail_a123demi_9837375]

参考文献：

王道远 《Spark  快速大数据分析》

[1240]: /images/20220616/38f687536ceb4cae8f1be549dd68fc65.png
[1240 1]: /images/20220616/4961f7039aff425bbe184ef8e69e8661.png
[1240 2]: /images/20220616/7caafa30f87a4a59a8229d736913cf7c.png
[http_download.csdn.net_detail_a123demi_9837375]: http://download.csdn.net/detail/a123demi/9837375