RDD 算子分类

摘要: 本文主要介绍Spark算子的作用，以及算子的分类。

转换：Transformation ， 行动： Action

RDD算子分类，大致可以分为两类，即：

1.  Transformation：转换算子，这类转换并不触发提交作业，完成作业中间过程处理。

2.  Action：行动算子，这类算子会触发SparkContext提交Job作业。

下面分别对两类算子进行详细介绍：

一：Transformation：转换算子

1.  map：

将原来RDD的每个数据项通过map中的用户自定义函数f映射转变为一个新的元素。源码中map算子相当于初始化一个RDD，新RDD叫做MappedRDD(this,sc.clean(f) )。即：

map是对RDD中的每个元素都执行一个指定的函数来产生一个新的RDD。任何原RDD中的元素在新RDD中都有且只有一个元素与之对应。

上述例子中把原RDD中每个元素都乘以3来产生一个新的RDD。

2.  mapPartitions：

mapPartitions函数获取到每个分区的迭代器，在函数中通过这个分区整体的迭代器对整个分区的元素进行 *** 作。内部实现是生成MapPartitionsRDD。

上述例子是通过 函数filter 对分区中所有数据进行过滤。

3.  mapValues

针对(key,value)型数据中的Value进行 *** 作，而不对Key进行处理。即：

mapValues顾名思义就是输入函数应用于RDD中Kev-Value的Value，原RDD中的Key保持不变，与新的Value一起组成新的RDD中的元素。因此，该函数只适用于元素为KV对的RDD。

4.  mapWith:

mapWith是map的另外一个变种，map只需要一个输入函数，而mapWith有两个输入函数。

eg: 把partition index 乘以10，然后加上2作为新的RDD的元素.(3 是将十个数分为三个区)

5.  flatMap：

将原来RDD中的每个元素通过函数f转换为新的元素，并将生成的RDD的每个集合中的元素合并为一个集合，内部创建FlatMappedRDD(this,sc.clean() )。即：

与map类似，区别是原RDD中的元素经map处理后只能生成一个元素，而原RDD中的元素经flatmap处理后可生成多个元素来构建新RDD。

eg：对原RDD中的每个元素x产生y个元素（从1到y，y为元素x的值）。

6.  flatMapWith：

flatMapWith与mapWith很类似，都是接收两个函数，一个函数把partitionIndex作为输入，输出是一个新类型A；另外一个函数是以二元组（T,A）作为输入，输出为一个序列，这些序列里面的元素组成了新的RDD。

7.  flatMapWithValues：

flatMapValues类似于mapValues，不同的在于flatMapValues应用于元素为KV对的RDD中Value。每个一元素的Value被输入函数映射为一系列的值，然后这些值再与原RDD中的Key组成一系列新的KV对。

上述例子中原RDD中每个元素的值被转换为一个序列（从其当前值到5），比如第一个KV对(1,2), 其值2被转换为2，3，4，5。然后其再与原KV对中Key组成一系列新的KV对(1,2),(1,3),(1,4),(1,5)。

8.  reduce：

reduce将RDD中元素两两传递给输入函数，同时产生一个新的值，新产生的值与RDD中下一个元素再被传递给输入函数直到最后只有一个值为止。

eg：对元素求和。

9.  reduceByKey

顾名思义，reduceByKey就是对元素为KV对的RDD中Key相同的元素的Value进行reduce，因此，Key相同的多个元素的值被reduce为一个值，然后与原RDD中的Key组成一个新的KV对。

eg：对Key相同的元素的值求和，因此Key为3的两个元素被转为了(3,10)。

10.  cartesian：

对两个RDD内的所有元素进行笛卡尔积 *** 作(耗内存)，内部实现返回CartesianRDD。

11.  Sample：

sample将RDD这个集合内的元素进行采样，获取所有元素的子集。用户可以设定是否有有放回的抽样，百分比，随机种子，进而决定采样方式。

内部实现： SampledRDD(withReplacement,fraction,seed)。

函数参数设置：

�             withReplacement=true，表示有放回的抽样。

�             withReplacement=false，表示无放回的抽样。

根据fraction指定的比例，对数据进行采样，可以选择是否用随机数进行替换，seed用于指定随机数生成器种子。

12.  union:

使用union函数时需要保证两个RDD元素的数据类型相同，返回的RDD数据类型和被合并的RDD元素数据类型相同。并不进行去重 *** 作，保存所有的元素，如果想去重，可以使用distinct()。同时，spark还提供更为简洁的使用union的API，即通过++符号相当于union函数 *** 作。

eg： a 与 b 的联合

去重复：

13.  groupBy：

将元素通过函数生成相应的Key，数据就转化为Key-Value格式，之后将Key相同的元素分为一组。

eg：根据数据集中的每个元素的K值对数据分组

14.  join：

join对两个需要连接的RDD进行cogroup函数 *** 作，将相同key的数据能偶放到一个分区，在cgroup *** 作之后形成新RDD对每个key下的元素进行笛卡尔积的 *** 作，返回的结果在展平，对应key下的所有元组形成一个集合。最后返回 RDD[(K， (V， W))]。

eg：a与b两个数据连接，相当于表的关联

15.  cache：

cache将RDD元素从磁盘缓存到内存。相当于 persist(MEMORY_ONLY) 函数的

功能。

16.  persist：

persist函数对RDD进行缓存 *** 作，数据缓存在哪里，由StorageLevel这个枚举类型进行确定。DISK 代表磁盘，MEMORY 代表内存， SER 代表数据是否进行序列化存储。

函数定义： persist(newLevel:StorageLevel)

StorageLevel 是枚举类型，代表存储模式。

MEMORY_AND_DISK_SER 代表数据可以存储在内存和磁盘，并且以序列化的方式存储，其他同理。

二：Action：行动算子

1.  foreach：

foreach对RDD中的每个元素都应用f函数 *** 作，不返回 RDD 和 Array， 而是返回Uint。

2.  saveAsTextFile：

函数将数据输出，存储到 HDFS 的指定目录。

函数的内 部实现，其内部通过调用 saveAsHadoopFile 进行实现：

this.map(x =>(NullWritable.get(), new Text(x.toString)))

.saveAsHadoopFile[TextOutputFormat[NullWritable, Text]](path)

将 RDD 中的每个元素映射转变为 (null， x.toString)，然后再将其写入 HDFS。

3.  collect：

collect相当于toArray，不过已经过时不推荐使用，collect将分布式的RDD返回为一个单机的scala Array数据，在这个数组上运用 scala 的函数式 *** 作。

4.  count：

count返回整个RDD的元素个数。

一般来讲,对于陌生的名词,大家的第一个反应都是“What is it?”.

RDD是Spark的核心内容,在Spark的官方文档中解释如下：RDD is a fault-tolerant collection of elements that can be operated on in parallel.由此可见,其中有两个关键词：fault-tolerant &in parallel.首先,容错性是RDD的一个重要特性；其次,它是并行计算的数据.

RDD的中文解释为：d性分布式数据集,全称Resilient Distributed Datasets.宾语是dataset,即内存中的数据库.RDD 只读、可分区,这个数据集的全部或部分可以缓存在内存中,在多次计算间重用.所谓d性,是指内存不够时可以与磁盘进行交换.这涉及到了RDD的另一特性：内存计算,就是将数据保存到内存中.同时,为解决内存容量限制问题,Spark为我们提供了最大的自由度,所有数据均可由我们来进行cache的设置,包括是否cache和如何cache.

如果看到这里,你的思维里对RDD还是没有任何概念的话,或许可以参照我的形象化理RDD,就是一个被武装起来的数据集.

在Spark编程中，常常需要对RDD进行修改、更新或者筛选等 *** 作，其中包括设为和增加两种 *** 作，这两种 *** 作的区别如下：

1. 设为 *** 作：将某个RDD替换为新的RDD，即将原有的RDD对象赋值为一个新的RDD对象，可以使用“=”符号或者“persist”方法来实现。设为 *** 作会将原有的RDD对象全部删除或者替换掉，所以在使用设为 *** 作时，需要注意对原有的RDD对象进行备份或者彻底删除。

2. 增加 *** 作：将某个RDD与新的RDD进行合并，即将新的RDD添加到原有的RDD中，可以使用“union”等方法来实现。增加 *** 作会保留原有的RDD对象不变，并将新的RDD对象添加进去，所以在使用增加 *** 作时，需要注意对数据的重复和重复计算问题。

总之，设为 *** 作和增加 *** 作都是RDD *** 作中常见的 *** 作，区别在于设为 *** 作会替换原有的RDD，而增加 *** 作会在原有的RDD基础上添加新的RDD。需要根据实际情况进行选择和使用。

---