阿帕奇火花移动平均线

Answer 1

移动平均线对于Spark和任何分布式系统来说都是一个棘手的问题。当数据分散在多台机器上时，会有一些跨分区的时间窗口。我们必须在分区开始时复制数据，这样计算每个分区的移动平均值就可以得到完全的覆盖。

这里有一种在Spark中做到这一点的方法。示例数据：

val ts = sc.parallelize(0 to 100, 10)
val window = 3

一个简单的分区器，它将每一行放在我们通过键指定的分区中：

class StraightPartitioner(p: Int) extends org.apache.spark.Partitioner {
  def numPartitions = p
  def getPartition(key: Any) = key.asInstanceOf[Int]
}

val partitioned = ts.mapPartitionsWithIndex((i, p) => {
  val overlap = p.take(window - 1).toArray
  val spill = overlap.iterator.map((i - 1, _))
  val keep = (overlap.iterator ++ p).map((i, _))
  if (i == 0) keep else keep ++ spill
}).partitionBy(new StraightPartitioner(ts.partitions.length)).values

val movingAverage = partitioned.mapPartitions(p => {
  val sorted = p.toSeq.sorted
  val olds = sorted.iterator
  val news = sorted.iterator
  var sum = news.take(window - 1).sum
  (olds zip news).map({ case (o, n) => {
    sum += n
    val v = sum
    sum -= o
    v
  }})
})

scala> movingAverage.collect.sameElements(3 to 297 by 3)
res0: Boolean = true