为什么在 Spark 中聚合和折叠两个不同的 API？答案

【问题标题】：Why are aggregate and fold two different APIs in Spark?为什么在 Spark 中聚合和折叠两个不同的 API？
【发布时间】：2014-10-29 15:47:05
【问题描述】：

当使用 Scala 标准库时，我可以这样做：

scala> val scalaList = List(1,2,3)
scalaList: List[Int] = List(1, 2, 3)

scala> scalaList.foldLeft(0)((acc,n)=>acc+n)
res0: Int = 6

从许多 Int 中制作一个 Int。

我可以这样做：

scala> scalaList.foldLeft("")((acc,n)=>acc+n.toString)
res1: String = 123

从多个 Int 中创建一个字符串。

因此，foldLeft 可以是同质的也可以是异构的，无论我们想要什么，它都在一个 API 中。

在 Spark 中，如果我想要多个 Int 中的一个 Int，我可以这样做：

scala> val rdd = sc.parallelize(List(1,2,3))
rdd: org.apache.spark.rdd.RDD[Int] = ParallelCollectionRDD[1] at parallelize at <console>:12
scala> rdd.fold(0)((acc,n)=>acc+n)
res1: Int = 6

fold API 与 foldLeft 类似，但它只是同构的，一个 RDD[Int] 只能产生带有 fold 的 Int。

spark 中也有一个聚合 API：

scala> rdd.aggregate("")((acc,n)=>acc+n.toString, (s1,s2)=>s1+s2)
res11: String = 132

它是异构的，一个RDD[Int]现在可以产生一个String。

那么，为什么在 Spark 中折叠和聚合实现为两个不同的 API？

为什么它们不像 foldLeft 那样设计，既可以是同质的，也可以是异构的？

（我是 Spark 的新手，如果这是一个愚蠢的问题，请原谅。）

【问题讨论】：

标签： scala aggregate apache-spark heterogeneous homogenous-transformation

【解决方案1】：

fold 可以更有效地实现，因为它不依赖于固定的评估顺序。所以每个集群节点可以并行fold它自己的块，然后在最后一个小的整体fold。而对于foldLeft，每个元素都必须按顺序折叠，并且不能并行执行任何操作。

（此外，为方便起见，为常见情况提供一个更简单的 API 也很好。标准库有 reduce 以及 foldLeft 出于这个原因）

【讨论】：

更准确地说，Spark 中的fold 需要关联性和交换性，而Scala 和其他非分布式框架中的fold 不需要交换性。查看 Spark 文档here

【解决方案2】：

特别是在 Spark 中，计算是分布式并行完成的，因此 foldLeft 无法像在标准库中那样实现。相反，聚合需要两个函数，一个对T 类型的每个元素执行类似于fold 的操作，生成U 类型的值，另一个将来自每个分区的U 组合到最终价值：

def aggregate[U: ClassTag](zeroValue: U)(seqOp: (U, T) => U, combOp: (U, U) => U): U

【讨论】：

【解决方案3】：

foldLeft, foldRight, reduceLeft, reduceRight, scanLeft和scanRight是累加参数可以不同于输入参数（(A, B) -> B）的操作，这些操作只能顺序执行。

fold 是累积参数必须与输入参数 ((A, A) -> A) 类型相同的操作。然后就可以并行执行了。

aggregation 是一种运算，其中累积的参数可以与输入参数的类型不同，但是您必须提供一个附加函数来定义如何将累积的参数聚合到最终结果中。此操作允许并行执行。 aggregation 操作是 foldLeft 和 fold 的组合。

如需更多详细信息，您可以观看“并行编程”课程的 coursera 视频：

【讨论】：