spark_RDD数据操作

RDD基础

RDD是Spark对数据的核心抽象—弹性分布式数据集（Resilient Distributed DataSet）。RDD表示分布在多个计算节点上不可变的、可以并行操作的元素集合。Spark中对RDD的操作包括创建RDD，转化已有的RDD（transformation）以及调用RDD操作（action）进行求值。

1.1 RDD的创建

? RDD的创建有两种方法：读取一个外部数据集，或在驱动器程序里分发驱动器程序的对象集合（比如list和set）

//读取外部数据集
val lines = sc.textFile("/path/to/README.txt")
//将已有的集合传给SparkContext的parallelize()方法
val lines = sc.parallelize(List("pandas","panpan"))

?

1.2 RDD的转化操作（transformation）

? 转换操作由一个RDD产生一个新的RDD。比如filter、map、join

? 转换操作都是惰性计算的，只有在第一行动操作中用到时，才会真正计算。

1.3 RDD的行动操作（action）

? 行动操作会对RDD计算出一个结果，并把结果返回到驱动器程序中，或者把结果写入外部存储系统（如HDFS）。

例如

val lines = sc.textFile("README.txt")
val pythonLines = lines.filter(line => line.contains("python"))
pythonLines.first()

? 第一行代码不会立即读取文本文件。当执行first（）行动操作时，驱动执行完成这个操作需要的RDD转化，此时Spark已经了解了完整的转化链，所以会自行优化操作，只计算求结果时真正用到的数据。在这个例子中，Spark只需要扫描文件找到第一个匹配的行为止，不需要读取整个文件。

? 默认情况下，RDD会在每次对它进行行动操作时重新计算。如果想要在多个行动操作中重用RDD，可以使用RDD.persist()将这个RDD**持久化**在内存（以分区的方式存储到集群中的各个机器上）里。

?

?

RDD特性

2.1 容错性

? 通过转化操作，从已有的RDD派生出新的RDD，Spark会使用谱系图（lineage graph）来记录这些不同RDD之间的依赖关系。我们不应该将RDD看作存放着特定数据的数据集，而最好把RDD当作我们通过转化操作构建出来的、记录如何计算数据的指令列表。Spark需要这些信息来按需计算每个RDD，也可以依靠谱系图在持久化的RDD丢失部分数据时恢复所丢失的数据。

2.2 分区优化性

? 在分布式程序中，通信的代价是很大的，因此控制数据分布以获得最小的网络传输可以极大地提升整体性能。Spark中的所有键值对RDD（pair RDD）可以进行分区（partitioniing），系统会根据一个针对键的函数对元素进行分组，确保同一组的键出现在同一个节点上。

? 例如join操作会将两个数据集中的所有键的哈希值都求出来，将该哈希值相同的记录通过网络传送到同一台机器上，然后在那台机器上对所有键相同的记录进行连接操作。对较大的、数据不会发生变化的数据每次进行跨节点的数据混洗（shuffle）会浪费很多时间，通过对这部分数据使用partitionBy转化操作将其表示为哈希分区，可以实现分区操作的本地化，减少网络传输的数据。

? Spark中有很多操作都引入了将数据根据键跨节点进行混洗的过程，所有这些操作都会从数据分区中获益。比如cogroup()、join()、groupWith()、leftOuterJoin()、rightOuterJoin()、groupByKey()、reduceByKey()等

2.3 适用性

? RDD适用于大数据环境下的批处理操作。在这种计算背景下,RDD可以高效地记住谱系图中的每一步的转化，在不备份大量数据的条件下重新计算出错的分区。它不适用于异步细粒度的共享状态更新应用，例如web应用的存储系统或者增量爬虫。

?

?

基础RDD常见的转化操作和行动操作

3.1转化操作

针对各个元素的操作：

map()：接受一个函数，将这个函数用于RDD中的每一个元素，将函数的返回结果作为结果RDD中对应元素的值

filter()：接受一个函数，将RDD中满足该函数的元素放入新的RDD中返回

flapMap()： 接受一个函数，将这个函数应用于RDD中的每一个元素。不过返回的不是一个元素，而是一个返回值序列的迭代器。输出的RDD倒不是由迭代器组成的。我们得到的是一个包含各个迭代器可访问的所有元素的RDD。例如把输入的字符串切分为单词。

伪集合操作

union()： 伪集合操作，缺少元素的唯一性。它会返回一个包含两个RDD中所有元素的RDD（包含重复数据）。例如处理多个数据源的日志文件。

intersection()：返回两个RDD钟都有的元素。性能较差，因为它需要网络混洗数据来发现共有的元素。

distinct()：去除重复数据。需要shuffle。

subtract(other)：接受 另一个RDD作为参数，返回一个只存在于第一个RDD而不存在于第二个RDD中的所有元素组成的RDD。需要shuffle.

3.2 行动操作

reduce()：接受一个函数作为参数，这个函数要操作两个相同数据类型的RDD数据并返回一个同样类型的新元素。

fold()：接受一个与reduce()接收的函数签名相同的函数，再加上一个“初始值”来作为每个分区第一次调用时的结果。你所提供的初始值应为你提供的操作的单位元素。

collect()：返回RDD中的所有元素（通常用于单元测试时数据集不是很大时）

take(num)：从RDD中返回num个元素。

foreach(func)：对RDD中的每个元素使用给定的函数。

以上的操作都针对于返回值类型与所操作的RDD类型相同的情况

aggregate()：与reduce()类似，但是通常返回不同类型的函数。

3.3 持久化操作:

persist()： 在scala和java里，默认情况下persist()会把数据以序列化的形式存储在JVM的堆空间里。可以指定持久化的级别。如果缓存总数据放不下，Spark会自动利用最近最少使用（LRU）的缓存策略将最老的分区从缓存中移除。

键值对RDD常见的转化操作和行动操作

键值对RDD通常用来进行聚合操作。很多键值对的数据格式会在读取时直接返回由其键值对数据组成的pair RDD。此外，当需要将一个普通RDD转换为pair RDD时，可以使用map函数实现。

4.1转化操作

subtractByKey(other)：删除RDD中键与other RDD中键相同的元素

mapValues()：对pair RDD中的每个值应用一个函数而不改变键

keys()：返回一个仅包含键的RDD

数据分组：

groupByKey()：对具有相同键的值进行分组

聚合操作：

reduceByKey()：规约每个键对应的数据

combineByKey(createCombiner,mergeValue,mergeCombiners,partitioner)：使用不同的返回类型合并具有相同键的值

连接操作：

join()：把两个RDD中键相同的元素组合在一起，合并为一个RDD

rightOuterJoin()：对两个RDD进行连接，确保第一个RDD的键必须存在。RDD1.rightOuterJoin(RDD2)

leftOuterJoin()：对两个RDD进行连接，确保第二个RDD的键必须存在。

cogroup()：将两个RDD中拥有相同键的数据分组。

数据排序：

sortByKey()：返回一个根据键排序的RDD

4.2 行动操作

包括所有基础RDD的行动操作

countByKey()：对每个键对应的元素分别计数

collectAsMap()：对结果以映射表的形式返回，以便查询

lookup()：返回给定键对应的所有值。

?

?

?

?

?

?