前言

RDD的基本操作分为两种,一种是转换Transformation,一种是行动Action


RDD中的所有转换都是延迟加载的,也就是说,它们并不会直接计算结果。相反的,它们只是记住这些应用到基础数据集(例如一个文件)上的转换动作。只有当发生一个要求返回结果给Driver的动作时,这些转换才会真正运行。这种设计让Spark更加有效率地运行。

各算子的详细代码示例,参考如下:
Spark算子使用示例 <https://blog.csdn.net/u013980127/article/details/53046760>

一、常用的Transformation

1)map(func)

返回一个新的RDD,该RDD由每一个输入元素经过func函数转换后组成

2)filter(func)

返回一个新的RDD,该RDD由经过func函数计算后返回值为true的输入元素组成

3)flatMap(func)

类似于map,但是每一个输入元素可以被映射为0或多个输出元素(所以func应该返回一个序列,而不是单一元素)

4)mapPartitions(func)

类似于map,但独立地在RDD的每一个分片上运行,因此在类型为T的RDD上运行时,func的函数类型必须是Iterator[T] =>
Iterator[U]。假设有N个元素,有M个分区,那么map的函数的将被调用N次,而mapPartitions被调用M次,一个函数一次处理所有分区

5)mapPartitionsWithIndex(func)

类似于mapPartitions,但func带有一个整数参数表示分片的索引值,因此在类型为T的RDD上运行时,func的函数类型必须是(Int,
Interator[T]) => Iterator[U]

6)sample(withReplacement, fraction, seed)


以指定的随机种子随机抽样出数量为fraction的数据,withReplacement表示是抽出的数据是否放回,true为有放回的抽样,false为无放回的抽样,seed用于指定随机数生成器种子。例子从RDD中随机且有放回的抽出50%的数据,随机种子值为3(即可能以1
2 3的其中一个起始值)

7)takeSample

和Sample的区别是:takeSample返回的是最终的结果集合。

8)intersection(otherDataset)

对源RDD和参数RDD求交集后返回一个新的RDD

9)distinct([numTasks]))

对源RDD进行去重后返回一个新的RDD. 默认情况下,只有8个并行任务来操作,但是可以传入一个可选的numTasks参数改变它。

10)partitionBy

对RDD进行分区操作,如果原有的partionRDD和现有的partionRDD是一致的话就不进行分区, 否则会生成ShuffleRDD。

11)reduceByKey(func, [numTasks])


在一个(K,V)的RDD上调用,返回一个(K,V)的RDD,使用指定的reduce函数,将相同key的值聚合到一起,reduce任务的个数可以通过第二个可选的参数来设置

12)groupByKey

groupByKey也是对每个key进行操作,但只生成一个sequence。

13)combineByKey[C]( createCombiner: V => C, mergeValue: (C, V) => C,
mergeCombiners: (C, C) => C)

对相同K,把V合并成一个集合.
createCombiner: combineByKey() 会遍历分区中的所有元素,因此每个元素的键要么还没有遇到过,要么就
和之前的某个元素的键相同。如果这是一个新的元素,combineByKey() 会使用一个叫作 createCombiner() 的函数来创建
那个键对应的累加器的初始值
mergeValue: 如果这是一个在处理当前分区之前已经遇到的键, 它会使用 mergeValue() 方法将该键的累加器对应的当前值与这个新的值进行合并
mergeCombiners: 由于每个分区都是独立处理的, 因此对于同一个键可以有多个累加器。如果有两个或者更多的分区都有对应同一个键的累加器,
就需要使用用户提供的 mergeCombiners() 方法将各个分区的结果进行合并。

14)aggregateByKey(zeroValue:U,[partitioner: Partitioner]) (seqOp: (U, V) =>
U,combOp: (U, U) => U)


在kv对的RDD中,,按key将value进行分组合并,合并时,将每个value和初始值作为seq函数的参数,进行计算,返回的结果作为一个新的kv对,然后再将结果按照key进行合并,最后将每个分组的value传递给combine函数进行计算(先将前两个value进行计算,将返回结果和下一个value传给combine函数,以此类推),将key与计算结果作为一个新的kv对输出。
seqOp函数用于在每一个分区中用初始值逐步迭代value,combOp函数用于合并每个分区中的结果

15)foldByKey(zeroValue: V)(func: (V, V) => V): RDD[(K, V)]

aggregateByKey的简化操作,seqop和combop相同

16)sortByKey([ascending], [numTasks])

在一个(K,V)的RDD上调用,K必须实现Ordered接口,返回一个按照key进行排序的(K,V)的RDD

17)sortBy(func,[ascending], [numTasks])

与sortByKey类似,但是更灵活,可以用func先对数据进行处理,按照处理后的数据比较结果排序。

18)join(otherDataset, [numTasks])

在类型为(K,V)和(K,W)的RDD上调用,返回一个相同key对应的所有元素对在一起的(K,(V,W))的RDD

19)cogroup(otherDataset, [numTasks])

在类型为(K,V)和(K,W)的RDD上调用,返回一个(K,(Iterable,Iterable))类型的RDD

20)cartesian(otherDataset)

笛卡尔积

21)pipe(command, [envVars])
对于每个分区,都执行一个perl或者shell脚本,返回输出的RDD

22)coalesce(numPartitions)

缩减分区数,用于大数据集过滤后,提高小数据集的执行效率。

23)repartition(numPartitions)

根据分区数,从新通过网络随机洗牌所有数据。

24)repartitionAndSortWithinPartitions(partitioner)


repartitionAndSortWithinPartitions函数是repartition函数的变种,与repartition函数不同的是,repartitionAndSortWithinPartitions在给定的partitioner内部进行排序,性能比repartition要高。

25)glom

将每一个分区形成一个数组,形成新的RDD类型时RDD[Array[T]]

26)mapValues

针对于(K,V)形式的类型只对V进行操作

27)subtract

计算差的一种函数去除两个RDD中相同的元素,不同的RDD将保留下来

二、常用的Action

1)reduce(func)

通过func函数聚集RDD中的所有元素,这个功能必须是可交换且可并联的

2)collect()

在驱动程序中,以数组的形式返回数据集的所有元素

3)count()

返回RDD的元素个数

4)first()

返回RDD的第一个元素(类似于take(1))

5)take(n)

返回一个由数据集的前n个元素组成的数组

6)takeSample(withReplacement,num, [seed])

返回一个数组,该数组由从数据集中随机采样的num个元素组成,可以选择是否用随机数替换不足的部分,seed用于指定随机数生成器种子

7)takeOrdered(n)

返回前几个的排序

8)aggregate (zeroValue: U)(seqOp: (U, T) ⇒ U, combOp: (U, U) ⇒ U)


aggregate函数将每个分区里面的元素通过seqOp和初始值进行聚合,然后用combine函数将每个分区的结果和初始值(zeroValue)进行combine操作。这个函数最终返回的类型不需要和RDD中元素类型一致。

9)fold(num)(func)

折叠操作,aggregate的简化操作,seqop和combop一样。

10)saveAsTextFile(path)


将数据集的元素以textfile的形式保存到HDFS文件系统或者其他支持的文件系统,对于每个元素,Spark将会调用toString方法,将它装换为文件中的文本

11)saveAsSequenceFile(path)

将数据集中的元素以Hadoop sequencefile的格式保存到指定的目录下,可以使HDFS或者其他Hadoop支持的文件系统。

12)saveAsObjectFile(path)

用于将RDD中的元素序列化成对象,存储到文件中。

13)countByKey()

针对(K,V)类型的RDD,返回一个(K,Int)的map,表示每一个key对应的元素个数。

14)foreach(func)

在数据集的每一个元素上,运行函数func进行更新。