Spark基础 --RDD算子详解

RDD算子分为两类：Transformation和Action，如下图，记住这张图，走遍天下都不怕。

Transformation：将一个RDD通过一种规则映射为另外一个RDD。

Action：返回结果或保存结果。

注意：只有action才触发程序的执行，transformation不触发执行。

RDD的操作种类有多个，分为： 单指RDD操作、Key/Value RDD操作、多个RDD联合操作，其他操作。

单值RDD

1. Map

map (f: T => U) : RDD[U] ，其中f定义了类型为T的元素到类型为U

的元素的映射，RDD[T] => RDD[U]的变换

举例：

  var rdd=sc.makeRDD(1 to 7,3)

简写为  rdd.map(_+1)   //rdd.map(x=>x+1)

2. collect

collect(): Array[T]，T是RDD中元素类型，将RDD转化为数组。

举例：

val rdd = sc.makeRDD(1 to 7, 3)

rdd.collect()

注意：此算子非常危险，他会将所有RDD中的数据汇总到Drive端的JVM内存中，对Drive端压力很大。

3. take

take(num: Int): Array[T] ，其中k是整数,T是RDD中元素类型，返回RDD中前k个元素，并保存成数组

举例：

val rdd = sc.makeRDD(1 to 7, 3)

rdd.take(2)

4. glom

glom() : RDD[Array[T]]，将RDD中每个partition中元素转换为数组

举例：

val rdd = sc.makeRDD(1 to 7, 3)

rdd.glom.collect

5. coalesce

coalesce(numPartitions: Int) : RDD[T]，将RDD中的partition个数合并为numPartitions个

举例：

val rdd = sc.makeRDD(1 to 7,7)

rdd.coalesce(3) // 生成新的RDD，它包含三个Partition

6. repartition

repartition(numPartitions: Int) :RDD[T]，将RDD中的partition个数均匀合并为numPartitions个

举例：

val list = Seq(Seq(),Seq(),Seq(),Seq(),Seq(),Seq(),

Seq(1,2,3,4,5,6,7))

val rdd = sc.makeRDD(list, 7).flatMap(x => x)

rdd.repartition(3) // 生成新的RDD，它包含三个Partition

7. filter

filter(f: T => Boolean):

RDD[T] ，其中f定义了类型为T的元素是否留下，过滤输入RDD中的元素，将f返回true的元素留下

举例：

var rdd=sc.makeRDD(1 to 7,7)

rdd.filter(_%3==0)

8. count

count(): Long,统计RDD中元素个数，并返回Long类型

val rdd = sc.makeRDD(1 to 7, 3)

rdd.count() // 统计RDD中元素总数

9. flatMap

flatMap(f: T =>TraversableOnce[U]): RDD[U],将函数f作用在RDD中每个元素上，并展开（flatten）

输出的每个结果, flatMap = flatten + map,先映射（map），再拍扁（flatten ）

举例：

val rdd = sc.makeRDD(1 to 3, 3)

rdd.flatMap( x => 1 to x) // 将x映射成1~x

10. reduce

reduce(f: (T, T) => T): T, 按照函数f对RDD中元素,进行规约

举例：

val rdd = sc.makeRDD(1 to 7, 3)

rdd.reduce((x, y) => x + y)

简写为：rdd.reduce(_ + _)

11. foreach

foreach(f: T => Unit):Unit,对RDD中每个元素，调用函数f

举例：

val rdd = sc.makeRDD(1 to 7, 3)

rdd.foreach( x => println(x))

简写为：rdd.foreach(println)

Key/Value RDD

首先先来看下如何创建一个Key/Value的rdd

var seq=Seq((A,1),(B,1),(C,1))

var rdd=sc.makeRDD(seq)

1. mapValues

对vaule做map操作

举例：

val pairs = Seq((A,1), (B,2), (A,2), (C, 4), (B, 1), (B, 1), (D, 1))

val rdd = sc.makeRDD(pairs, 3)

rdd.mapValues(_ + 1)

2. reduceByKey

对Key相同的value做计算

举例：

val pairs = Seq((‘A‘,1), (‘B‘,2), (‘A‘,2), (‘C‘, 4), (‘B‘, 1), (‘B‘, 1), (‘D‘, 1))

val rdd = sc.makeRDD(pairs, 3)

rdd.reduceByKey(_ + _)

3. groupByKey

将RDD[key,value] 按照相同的key进行分组，形成RDD[key,Iterable[value]]的形式， 有点类似于sql中的groupby

举例：

val pairs = Seq((A,1), (B,2), (A,2), (C, 4), (B, 1), (B, 1), (D, 1))

val rdd = sc.makeRDD(pairs, 3)

rdd.groupByKey()

注意：能用reducebykey代替就不用groupbykey，groupbykey会将所有的元素进行聚合，消耗大量内存。

多RDD

1. union

将多个RDD合并为一个RDD

举例：

val pairs1 = Seq((A,1), (B,1), (C,1), (D, 1), (A, 2), (C, 3))

val rdd1 = sc.makeRDD(pairs1, 3)

val pairs2 = Seq((A,4), (D,1), (E, 1))

val rdd2 = sc.makeRDD(pairs2, 2)

rdd1.union(rdd2)

2. zip

zip函数用于将两个RDD组合成Key/Value形式的RDD，如果两个rdd中的partition数量不一致，会报错。

举例：

val s1 = Seq(A, B, C, D, E)

val rdd1 = sc.makeRDD(s1)

val s2 = Seq(1, 2, 3, 4, 5)

val rdd2 = sc.makeRDD(s2)

rdd1.zip(rdd2)

3. join

join相当于SQL中的内关联join，只返回两个RDD根据K可以关联上的结果，join只能用于两个RDD之间的关联，

如果要多个RDD关联，多关联几次即可

举例：

val pairs1 = Seq((A,1), (B,1), (C,1), (D, 1), (A, 2), (C, 3))

val rdd1 = sc.makeRDD(pairs1, 3)

val pairs2 = Seq((A,4), (D,1), (C,1), (E, 1))

val rdd2 = sc.makeRDD(pairs2, 2)

rdd1.join(rdd2)

还有些是是其他rdd操作符，这里就不讲解了，上述所写如有不对之处，还请各位前辈赐教。

原文地址：https://www.cnblogs.com/wuweikongjian/p/8309245.html