在 Scala 中聚合 Spark 数据集的问题答案

【问题标题】：Issues Aggregating Spark Datasets in Scala在 Scala 中聚合 Spark 数据集的问题
【发布时间】：2017-05-26 23:25:10
【问题描述】：

我正在使用 scala 的 /: 运算符计算一系列数据集聚合。下面列出了聚合的代码：

def execute1( 
xy: DATASET, 
f: Double => Double): Double = {

println("PRINTING: The data points being evaluated: " + xy)
println("PRINTING: Running execute1")

var z = xy.filter{ case(x, y) => abs(y) > EPS}

var ret = - z./:(0.0) { case(s, (x, y)) => {
   var px = f(x)
   s + px*log(px/y)}  
}

ret
}

当我尝试运行作为 f 参数传入的单独函数列表的块时，会出现我的问题。函数列表是：

  lazy val pdfs = Map[Int, Double => Double](
1 -> betaScaled,
2 -> gammaScaled,
3 -> logNormal,
4 -> uniform,
5 -> chiSquaredScaled
)

通过列表运行聚合的执行器函数是：

  def execute2( 
xy: DATASET, 
fs: Iterable[Double=>Double]): Iterable[Double] = { 
fs.map(execute1(xy, _))
}

最后的执行块：

val kl_rdd  = master_ds.mapPartitions((it:DATASET) => {
val pdfsList = pdfs_broadcast.value.map(
     n => pdfs.get(n).get
)

execute2(it, pdfsList).iterator

问题是，虽然确实发生了聚合，但它们似乎都聚合在输出数组的第一个槽中，而我希望单独显示每个函数的聚合。我进行了测试以确认所有五个函数实际上都在运行，并且它们在第一个槽中被求和。

The pre-divergence value: -4.999635700491883
The pre-divergence value: -0.0
The pre-divergence value: -0.0
The pre-divergence value: -0.0
The pre-divergence value: -0.0

这是我遇到过的最困难的问题之一，因此任何方向都将不胜感激。将给予应得的信用。谢谢！

【问题讨论】：

标签： scala apache-spark apache-spark-dataset

【解决方案1】：

Spark 的数据集没有foldLeft（又名/:）：https://spark.apache.org/docs/2.1.0/api/scala/index.html#org.apache.spark.sql.Dataset，实际上需要类型参数DataSet[T]，并且它的名称并非全部大写。

所以，我想你的DATASET 的类型是一个迭代器，所以在第一次运行execute1 后它会被耗尽，所以每个后续的execute1 都会得到空迭代器。基本上，它不会聚合所有函数 - 它只是执行第一个函数并忽略其他函数（你得到 -0.0，因为你将 0.0 作为初始值传递给 foldLeft）。

从mapPartitions签名可以看出：

def mapPartitions[U](f: (Iterator[T]) ⇒ Iterator[U], preservesPartitioning: Boolean = false)(implicit arg0: ClassTag[U]): RDD[U]

它为您提供了一个迭代器（只能遍历一次的可变结构），因此您应该执行it.toList 以获得（可能但有限的大）不可变结构（List）。

附：如果您想真正使用 Spark 的 DataSet/RDD - 使用 aggregate (RDD) 或 agg (DataSet)。另见：foldLeft or foldRight equivalent in Spark?

关于迭代器的解释：

scala> val it = List(1,2,3).toIterator
it: Iterator[Int] = non-empty iterator

scala> it.toList //traverse iterator and accumulate its data into List
res0: List[Int] = List(1, 2, 3)

scala> it.toList //iterator is drained, so second call doesn't traverse anything
res1: List[Int] = List()

【讨论】：