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

【问题标题】：Calculate average over RDD[Vector] in Spark在 Spark 中计算 RDD[Vector] 的平均值
【发布时间】：2015-11-26 08:23:51
【问题描述】：

我有一个 Breeze Vectors 的 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://stackoverflow.com/a/1346890/1037094）。

如何在 Spark 中为 Breeze Vectors 实现 Knuths algorithm 并且推荐使用 rdd.aggregate 方法？

【问题讨论】：

标签： scala apache-spark apache-spark-mllib

【解决方案1】：

如何在 Spark 中为 Breeze Vectors 实现 Knuths 算法，rdd.aggregate 是推荐的方法吗？

aggregate 可能是一个很好的方法，如果 Knuth 描述的算法是正确的选择。不幸的是，它不是，或者至少在没有一些调整的情况下不会。它本质上是顺序流算法，它应用的函数不是关联的。假设您有一个函数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 方法查看它在 Spark 内部是如何处理的。

【讨论】：