Spark API 全集(3):Spark RDD API全集

Posted on Edited on

RDD是啥

Resilient Distributed Dataset (RDD)，弹性分布式数据集，是对不可修改，分区的数据集合的抽象。

RDD is characterized by five main properties:

  • A list of partitions
  • A function for computing each split
  • A list of dependencies on other RDDs
  • Optionally, a Partitioner for key-value RDDs (e.g. to say that the RDD is hash-partitioned)
  • Optionally, a list of preferred locations to compute each split on (e.g. block locations for an HDFS file)

org.spark.rdd.RDD类方法

RDD是一个抽象类，定义如下

1
abstract class RDD[T] extends Serializable with Logging

RDD类的public方法大约有80多个（包括不同参数重载的）,均在下面列出。

值得注意的是，RDD类中并没有定义xxxByKey形式的方法，这类方法其实是在PairRDDFunctions中定义的，通过隐式转换，键值对形式的RDD（即RDD[(K, V)）可以调用PairRDDFunctions中定义的方法。相关的隐式转换定义在RDD的伴生对象中。

键值转换操作

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
filter(f: (T) ⇒ Boolean): RDD[T]
过滤数据，仅留下使得f返回true的元素。

map[U](f: (T) ⇒ U)(implicit arg0: ClassTag[U]): RDD[U]
将一个RDD中的每个数据项，通过map中的函数映射变为一个新的元素。
输入分区与输出分区一对一，即：有多少个输入分区，就有多少个输出分区。

flatMap[U](f: (T) ⇒ TraversableOnce[U])(implicit arg0: ClassTag[U]): RDD[U]
第一步和map一样，最后将所有的输出分区合并成一个。
使用flatMap时候需要注意：
flatMap会将字符串看成是一个字符数组。

mapPartitions[U](f: (Iterator[T]) ⇒ Iterator[U], preservesPartitioning: Boolean = false)(implicit arg0: ClassTag[U]): RDD[U]
该函数和map函数类似，只不过映射函数的参数由RDD中的每一个元素变成了RDD中每一个分区的迭代器。如果在映射的过程中需要频繁创建额外的对象，使用mapPartitions要比map高效的过。
比如，将RDD中的所有数据通过JDBC连接写入数据库，如果使用map函数，可能要为每一个元素都创建一个connection，这样开销很大，如果使用mapPartitions，那么只需要针对每一个分区建立一个connection。
参数preservesPartitioning表示是否保留父RDD的partitioner分区信息。

mapPartitionsWithIndex[U](f: (Int, Iterator[T]) => Iterator[U], preservesPartitioning: Boolean = false)(implicit arg0: ClassTag[U]): RDD[U]
函数作用同mapPartitions，不过提供了两个参数，第一个参数为分区的索引。

keyBy[K](f: (T) ⇒ K): RDD[(K, T)]
通过f函数为每个元素生成一个KEY

sortBy[K](f: (T) ⇒ K, ascending: Boolean = true, numPartitions: Int = this.partitions.length)(implicit ord: Ordering[K], ctag: ClassTag[K]): RDD[T]
通过给定的函数对元素排序

zip[U](other: RDD[U])(implicit arg0: ClassTag[U]): RDD[(T, U)]
与另一个RDD组合成（k,v)对。

zipPartitions[B, V](rdd2: RDD[B], preservesPartitioning: Boolean)(f: (Iterator[T], Iterator[B]) ⇒ Iterator[V])(implicit arg0: ClassTag[B], arg1: ClassTag[V]): RDD[V]

zipWithIndex(): RDD[(T, Long)]

zipWithUniqueId(): RDD[(T, Long)]


聚合相关

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
aggregate[U](zeroValue: U)(seqOp: (U, T) ⇒ U, combOp: (U, U) ⇒ U)(implicit arg0: ClassTag[U]): U
aggregate用户聚合RDD中的元素，先使用seqOp将RDD中每个分区中的T类型元素聚合成U类型，再使用combOp将之前每个分区聚合后的U类型聚合成U类型，特别注意seqOp和combOp都会使用zeroValue的值，zeroValue的类型为U。

treeAggregate[U](zeroValue: U)(seqOp: (U, T) ⇒ U, combOp: (U, U) ⇒ U, depth: Int = 2)(implicit arg0: ClassTag[U]): U
多层级聚合

reduce(f: (T, T) ⇒ T): T
根据映射函数f，对RDD中的元素进行二元计算，返回计算结果。

treeReduce(f: (T, T) ⇒ T, depth: Int = 2): T
多级reduce归并聚合

fold(zeroValue: T)(op: (T, T) ⇒ T): T
fold是aggregate的简化，将aggregate中的seqOp和combOp使用同一个函数op。


count(): Long
count返回RDD中的元素数量。

countApprox(timeout: Long, confidence: Double = 0.95): PartialResult[BoundedDouble]
近似count

countApproxDistinct(relativeSD: Double = 0.05): Long
countApproxDistinct(p: Int, sp: Int): Long
近似distinct count

countByValue()(implicit ord: Ordering[T] = null): Map[T, Long]
计算每个值出现次数

countByValueApprox(timeout: Long, confidence: Double = 0.95)(implicit ord: Ordering[T] = null):
计算每个值出现次数近似值

distinct(): RDD[T]
distinct(numPartitions: Int)(implicit ord: Ordering[T] = null): RDD[T]
返回元素去重后的RDD

groupBy[K](f: (T) ⇒ K)(implicit kt: ClassTag[K]): RDD[(K, Iterable[T])]
groupBy[K](f: (T) ⇒ K, numPartitions: Int)(implicit kt: ClassTag[K]): RDD[(K, Iterable[T])]
groupBy[K](f: (T) ⇒ K, p: Partitioner)(implicit kt: ClassTag[K], ord: Ordering[K] = null): RDD[(K, Iterable[T])]
按指定函数生成key，并按key分组。
注意：性能比较差，推荐用PairRDDFunctions.reduceByKey or PairRDDFunctions.aggregateByKey.
因为reduceByKey会先在分区内做聚合，再进行数据交换(shuffle)。

glom(): RDD[Array[T]]
该函数是将RDD中每一个分区中类型为T的元素转换成Array[T]，这样每一个分区就只有一个数组元素。

max()(implicit ord: Ordering[T]): T
最大的元素

min()(implicit ord: Ordering[T]): T
最小的元素


遍历元素

1
2
3
4
5
6
7
8
9
foreach(f: (T) ⇒ Unit): Unit
foreach用于遍历RDD,将函数f应用于每一个元素。
但要注意，如果对RDD执行foreach，只会在Executor端有效，而并不是Driver端。
比如：rdd.foreach(println)，只会在Executor的stdout中打印出来，Driver端是看不到的。

foreachPartition(f: (Iterator[T]) ⇒ Unit): Unit
foreachPartition和foreach类似，只不过是对每一个分区使用f。


取元素相关

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
collect(): Array[T]
collect用于将一个RDD转换成数组。

first(): T
first返回RDD中的第一个元素，不排序。

take(num: Int): Array[T]
take用于获取RDD中从0到num-1下标的元素，不排序。

top(num: Int)(implicit ord: Ordering[T]): Array[T]
top函数用于从RDD中，按照默认（降序）或者指定的排序规则，返回前num个元素。

takeOrdered(num: Int)(implicit ord: Ordering[T]): Array[T]
takeOrdered和top类似，只不过以和top相反的顺序返回元素

takeSample(withReplacement: Boolean, num: Int, seed: Long = Utils.random.nextLong): Array[T]
取样本元素



集合间运算

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19

++(other: RDD[T]): RDD[T]
与另一个RDD union。

intersection(other: RDD[T], partitioner: Partitioner)(implicit ord: Ordering[T] = null): RDD[T]
intersection(other: RDD[T], numPartitions: Int): RDD[T]
intersection(other: RDD[T]): RDD[T]
取交集

subtract(other: RDD[T], p: Partitioner)(implicit ord: Ordering[T] = null): RDD[T]
subtract(other: RDD[T], numPartitions: Int): RDD[T]
subtract(other: RDD[T]): RDD[T]
求差集

union(other: RDD[T]): RDD[T]
与另一个RDD合并，类似union all,不会去重。



其他

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
persist(): RDD.this.type
persist(newLevel: StorageLevel): RDD.this.type
缓存数据，可设置缓存级别(如果尚未设置过，才可以设置，本地checkpoint除外)

unpersist(blocking: Boolean = true): RDD.this.type
Mark the RDD as non-persistent, and remove all blocks for it from memory and disk.


cache(): RDD.this.type
MEMORY_ONLY级别缓存数据

cartesian[U](other: RDD[U])(implicit arg0: ClassTag[U]): RDD[(T, U)]：
计算两个RDD的迪卡尔积

checkpoint(): Unit
标记将该RDD进行checkpoint处理？

coalesce(numPartitions: Int, shuffle: Boolean = false)(implicit ord: Ordering[T] = null): RDD[T]
分区合并(只能减少分区)，使用HashPartitioner。
第一个参数为重分区的数目，第二个为是否进行shuffle，默认为false;

repartition(numPartitions: Int)(implicit ord: Ordering[T] = null): RDD[T]
调整分区数，会导致shuffle，如果是减少分区，可以使用coalesce，避免shuffle。

toDebugString: String
返回RDD依赖树/血统图

getCheckpointFile: Option[String]
获取checkpoint文件夹名称

localCheckpoint(): RDD.this.type
标记为使用本地checkpoint

isEmpty(): Boolean
是否含0个元素

iterator(split: Partition, context: TaskContext): Iterator[T]
返回迭代器，不应直接调用，而是给RDD的子类用的。

toLocalIterator: Iterator[T]
返回元素的本地迭代器

pipe(command: String): RDD[String]
pipe(command: String, env: Map[String, String]): RDD[String]
pipe(command: Seq[String], env: Map[String, String] = Map(), printPipeContext: ((String) ⇒ Unit) ⇒ Unit = null, printRDDElement: (T, (String) ⇒ Unit) ⇒ Unit = null, separateWorkingDir: Boolean = false, bufferSize: Int = 8192, encoding: String = Codec.defaultCharsetCodec.name): RDD[String]
调用外部进程处理RDD,如通过标准输入传给shell脚本。

preferredLocations(split: Partition): Seq[String]
Get the preferred locations of a partition, taking into account whether the RDD is checkpointed.

randomSplit(weights: Array[Double], seed: Long = Utils.random.nextLong): Array[RDD[T]]
按权随机将元素分组

sample(withReplacement: Boolean, fraction: Double, seed: Long = Utils.random.nextLong): RDD[T]
取样本/子集

setName(_name: String): RDD.this.type
设置RDD名字

保存

1
2
3
4
5
6
7
8
saveAsObjectFile(path: String): Unit
保存为SequenceFile

saveAsTextFile(path: String, codec: Class[_ <: CompressionCodec]): Unit
saveAsTextFile(path: String): Unit
保存为文本文件


变量

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
context: SparkContext
创建RDD的SparkContext

sparkContext: SparkContext
创建RDD的SparkContext

dependencies: Seq[Dependency[_]]
RDD的依赖列表

getNumPartitions: Int
获取RDD的分区数

getStorageLevel: StorageLevel
获取存储等级，如果设置为none,则返回StorageLevel.NONE 。

id: Int
该RDD的unique ID

isCheckpointed: Boolean
是否checkpointed and materialized, either reliably or locally.

name: String
RDD的名字

partitioner: Option[Partitioner]
分区器

partitions: Array[Partition]
各个分区