在Spark中计算RDD[Vector]的平均值

Question

我有一组Breeze向量的RDD，想要计算它们的平均值。我的第一种方法是使用aggregate：

import org.apache.spark.{ SparkConf, SparkContext }
import org.apache.spark.rdd.RDD
import org.scalatest.{ BeforeAndAfterAll, FunSuite, Matchers, Suite }
import org.scalatest.prop.GeneratorDrivenPropertyChecks

import breeze.linalg.{ Vector => BreezeVector }

class CalculateMean extends FunSuite with Matchers with GeneratorDrivenPropertyChecks with SparkSpec {

  test("Calculate mean") {

    type U = (BreezeVector[Double], Int)
    type T = BreezeVector[Double]
    val rdd: RDD[T] = sc.parallelize(List(1.0, 2, 3, 4, 5, 6).map { x => BreezeVector(x, x * x) }, 2)

    val zeroValue = (BreezeVector.zeros[Double](2), 0)
    val seqOp = (agg: U, x: T) => (agg._1 + x, agg._2 + 1)
    val combOp = (xs: U, ys: U) => (xs._1 + ys._1, xs._2 + ys._2)

    val mean = rdd.aggregate(zeroValue)(seqOp, combOp)
    println(mean._1 / mean._2.toDouble)

  }

}

/**
 * Setup and tear down spark context
 */
trait SparkSpec extends BeforeAndAfterAll {
  this: Suite =>

  private val master = "local[2]"
  private val appName = this.getClass.getSimpleName

  private var _sc: SparkContext = _

  def sc: org.apache.spark.SparkContext = _sc

  val conf: SparkConf = new SparkConf()
    .setMaster(master)
    .setAppName(appName)

  override def beforeAll(): Unit = {
    super.beforeAll()
    _sc = new SparkContext(conf)
  }

  override def afterAll(): Unit = {
    if (_sc != null) {
      _sc.stop()
      _sc = null
    }

    super.afterAll()
  }
}

然而，这个算法可能是数值不稳定的（参见 https://dev59.com/mXM_5IYBdhLWcg3wgzdl#1346890）。

我如何在Spark中为Breeze向量实现Knuths算法，并且rdd.aggregate是否是推荐的方法？

- Manuel Schmidt

Answer 1

我该如何在Spark中实现Knuth的Breeze向量算法？rdd.aggregate是推荐的方法吗？如果Knuth描述的算法是正确的选择，那么aggregate可能是一个不错的方式。但不幸的是，它并不是，或者至少需要进行一些调整。它是一个固有的顺序流算法，并且应用的函数不是可结合的。假设您有一个名为knuth_mean的函数。显然，（忽略计数和单个元素情况），

(knuth_mean (knuth_mean (knuth_mean 1 2) 3) 4)

它不同于

(knuth_mean (knuth_mean 1 2) (knuth_mean 3 4))

然而，您可以使用Knuth算法来获得每个分区的平均值：

def partMean(n: Int)(iter: Iterator[BreezeVector[Double]]) = {
  val partialMean = iter.foldLeft((BreezeVector.zeros[Double](n), 0.0))(
    (acc: (BreezeVector[Double], Double), v: BreezeVector[Double]) => 
      (acc._1 + (v - acc._1) / (acc._2 + 1.0), acc._2 + 1.0))
    Iterator(partialMean)
}

val means = rdd.mapPartitions(partMean(lengthOfVector))

问题仍然是如何聚合这些部分结果。直接应用 Knuth 算法需要展开分区，这几乎打败了使用 Spark 的整个目的。您可以使用 StatCounter.merge 方法来查看内部如何处理它。

- zero323