Spark算子 - fold()

 

算子定义： def fold(zeroValue: T)(op: (T, T) ⇒ T): T

 

1. 先看看文档中的定义:

Aggregate the elements of each partition, and then the results for all the partitions, using a given associative function and a neutral "zero value". The function op(t1, t2) is allowed to modify t1 and return it as its result value to avoid object allocation; however, it should not modify t2.

意思是说：fold 有俩参数，一个是初始值 zeroValue ，另一个是 op(t1, t2)方法。 在运算过程中， 会先对每个分区里的元素做聚合， 再对所有的分区结果做聚合；  op(t1, t2) 这个方法允许修改并返回t1的值来避免对象分配， 但不应该改变t2的值；

 

2. 再看看俩参数的详细解释:

zeroValue

the initial value for the accumulated result of each partition for the op operator, and also the initial value for the combine results from different partitions for the op operator - this will typically be the neutral element (e.g. Nil for list concatenation or 0 for summation)

译：这个初始值， 它既是op方法每个分区里累加器的初始值， 也是对不同分区结果进行合并的初始值，  它通常是个中性值(例如，列表为Nil，求和为0)

op

an operator used to both accumulate results within a partition and combine results from different partitions

译: 一个用来在分区内和分区之间进行结果累积的操作；

 

3. 最后结合例子来进一步理解它的意思。

   该例子对 (1, 2, 3, 4, 5, 6) 进行求和， 初始值为10,  分别用1个分区、2个分区、3个分区来观察最终结果:

'''

def localFold():Any = {
  val sourceRDD1 = sc.parallelize(Array(1,2,3,4,5,6), 1)
  println("1 partition:", sourceRDD1.fold(10)((x,y) => x+y))
  val sourceRDD2 = sourceRDD1.repartition(2)
  println("2 partition:", sourceRDD2.fold(10)((x,y) => x+y))
  val sourceRDD3 = sourceRDD1.repartition(3)
  println("3 partition:", sourceRDD3.fold(10)((x,y) => x+y))
}

(1 partition:,41)

(2 partition:,51)

(3 partition:,61)

'''

让我们来详细分析一下它的步骤：

参照上面译文: 

 当有一个分区时：

 第一步： 先对分区内元素做op操作， zeroValue作为初始值,  即:    10+1+2+3+4+5+6 = 31;

 第二步： 再对多个分区做op操作, zeroValue也作为初始值,  即 10 + 31 = 41;

 

当有两个分区时： (此处假设 partition0 (1, 2, 3),  partition1(4, 5, 6)  )

第一步： 先对分区内元素做op操作， zeroValue作为初始值, 

      partition0:  10+1+2+3 = 16

      partition1:  10+4+5+6 = 25

 第二步： 再对多个分区做op操作, zeroValue也作为初始值,  即 10 + 16 + 25 = 51;

 

若此文对你有所帮助， 请留个赞吧。。。

 

---- The END ----