请问如何解释这段代码并输出 (10,4)?
使用聚合函数时需要提供三个参数:
1. 从分区中累加元素的初始值,通常是中性元素。
2. 给定分区的函数将在其中累积结果。
3. 将两个分区组合的函数。
在您的情况下,分区的初始值是元组(0, 0)。
然后,您定义的累加器函数将当前遍历的元素与元组的第一个元素相加,并使元组的第二个元素加1。事实上,它将计算分区中元素的总和及其元素数量。
组合函数将两个元组组合在一起。根据您的定义,它将对2个分区的总和和元素数量进行求和。因为您按顺序遍历了管道,所以这不会被使用。您可以在List上调用“.par”以获得并行实现,从而看到组合器的操作(注意它必须是可结合的函数)。
因此,您得到(10, 4),因为1+2+3+4=10,在列表中有4个元素(您做了4次加法)。
您可以在累加器函数中添加打印语句(在顺序输入上运行),以查看其行为方式。
Acc: (0,0) - value:1
Acc: (1,1) - value:2
Acc: (3,2) - value:3
Acc: (6,3) - value:4
我知道scala中普通聚合是如何工作的以及它在fold上的应用。
对于顺序输入,
aggregate
就是一个
foldLeft
:
def aggregate[B](z: =>B)(seqop: (B, A) => B, combop: (B, B) => B): B = foldLeft(z)(seqop)
对于并行输入,列表将被分成多个块,以便多个线程可以分别处理。累加器函数在每个块上运行,使用初始值。当两个线程需要合并其结果时,使用组合函数:
def aggregate[S](z: =>S)(seqop: (S, T) => S, combop: (S, S) => S): S = {
tasksupport.executeAndWaitResult(new Aggregate(() => z, seqop, combop, splitter))
}
这是fork-join模型的原理,但它要求你的任务可以很好地并行化。在这里,这种情况是成立的,因为一个线程不需要知道另一个线程的结果就可以完成其工作。