Spark1.4.1 RDD算子详解

Spark 常用算子实现原理详解

Spark 常用算子实现原理详解...............................................................................................................................................1

1. take(num:Int)...............................................................................................................................................................2

2. first()............................................................................................................................................................................4

3. sortByKey(ascending: Boolean = true, numPartitions: Int = self.partitions.length)..................................................4

4. count()........................................................................................................................................................................13

5. countApprox( timeout: Long, confidence: Double = 0.95)......................................................................................13

6. countApproxDistinct(relativeSD: Double = 0.05)....................................................................................................17

13. reduce(f: (T, T) => T)..............................................................................................................................................25

14. max()........................................................................................................................................................................26

15. min().........................................................................................................................................................................27

16. treeReduce(f: (T, T) => T).......................................................................................................................................27

17. map[U: ClassTag](f: T => U)..................................................................................................................................28

18. mapPartitions[U: ClassTag]( f: Iterator[T] => Iterator[U], preservesPartitioning: Boolean = false).....................29

19. mapPartitionsWithIndex[U: ClassTag]( f: (Int, Iterator[T]) => Iterator[U], preservesPartitioning: Boolean =

false)...............................................................................................................................................................................30

20. flatMap[U: ClassTag](f: T => TraversableOnce[U])..............................................................................................31

21. filter(f: T => Boolean).............................................................................................................................................31

27. repartition(numPartitions: Int).................................................................................................................................51

28. sample( withReplacement: Boolean, fraction: Double, seed: Long = Utils.random.nextLong)............................52

29. takeSample( withReplacement: Boolean, num: Int, seed: Long = Utils.random.nextLong)..................................54

30. randomSplit( weights: Array[Double], seed: Long = Utils.random.nextLong)......................................................56

31. union(other: RDD[T])..............................................................................................................................................58

32. ++(other: RDD[T])..................................................................................................................................................63

33. intersection(other: RDD[T])....................................................................................................................................63

34. glom.........................................................................................................................................................................68

35. cartesian[U: ClassTag].............................................................................................................................................69

36. zip[U: ClassTag](other: RDD[U])...........................................................................................................................70

43. keyBy[K] (f: T => K)..............................................................................................................................................81

1. take(num:Int)

获取前 num 条记录。

def take(num: Int): Array[T] = withScope {

if (num == 0) {

new Array[T](0)

} else {

val buf = new ArrayBuffer[T]

val totalParts = this.partitions.length

var partsScanned = 0

// Otherwise, interpolate the number of partitions we need to try, but overestimate

// it by 50%. We also cap the estimation in the end.

if (buf.size == 0) {//截止目前为止 buf 为空的话，则扩大 4 倍范围

numPartsToTry = partsScanned * 4

} else {//截止目前为止还有部分值没取到的话，则扩大至 Math.max((1.5 * num * partsScanned / buf.size).toInt

- partsScanned, 1)，但是不超过当前已扫描过分区的 4 倍

// the left side of max is >=1 whenever partsScanned >= 2

numPartsToTry = Math.max((1.5 * num * partsScanned / buf.size).toInt - partsScanned, 1)

numPartsToTry = Math.min(numPartsToTry, partsScanned * 4)

}

}

buf.toArray

}

}

首先关注下 sc.runJob 函数的传参：

/**

* Run a job on a given set of partitions of an RDD, but take a function of type

* `Iterator[T] => U` instead of `(TaskContext, Iterator[T]) => U`.

*/

def runJob[T, U: ClassTag](

runJob(rdd, (ctx: TaskContext, it: Iterator[T]) => cleanedFunc(it), partitions, allowLocal)

}

其中 partitions: Seq[Int]代表需要计算的分区，可以计算某个分区，也可以计算多个分区，是待计算的分区集合。

其次先看第一次循环，其 partsScanned 为 0，numPartsToTry 为 1，因此先计算第一个分区的结果，如果第一次计

算可以取得满足条件的 num 个值，则循环结束，如果取不到满足条件的 num 个值，则扩大第二次计算的分区范

围，很可能一下子扫多个分区。

其执行过程见下图：

Take 可以避免全量计算，执行时间比较短。

2. first()

取 RDD 的第一个元素

case _ => throw new UnsupportedOperationException("empty collection")

}

}

其实就是调用 take 来完成的，take 的流程可以查阅 take 函数详解

3. sortByKey(ascending: Boolean = true, numPartitions: Int =

self.partitions.length)

def sortByKey(ascending: Boolean = true, numPartitions: Int = self.partitions.length)

: RDD[(K, V)] = self.withScope

{

val part = new RangePartitioner(numPartitions, self, ascending)

ShuffledRDD 拉取自己对应分区的数据。但是 sortByKey 主要应该掌握其 RangePartitioner 分区函数的执行原理，

它如何保证 ShuffledRDD 的每个分区的数量是大致相同的，也就是如何来划分每个分区的边界的，且看：

class RangePartitioner[K : Ordering : ClassTag, V](

@transient partitions: Int,

@transient rdd: RDD[_ <: Product2[K, V]],

private var ascending: Boolean = true)

extends Partitioner {

// We allow partitions = 0, which happens when sorting an empty RDD under the default settings.

require(partitions >= 0, s"Number of partitions cannot be negative but found $partitions.")

private var ordering = implicitly[Ordering[K]]

val sampleSize = math.min(20.0 * partitions, 1e6)

// Assume the input partitions are roughly balanced and over-sample a little bit.

val sampleSizePerPartition = math.ceil(3.0 * sampleSize / rdd.partitions.size).toInt

// numItems

rdd

// sketched