如何构建一个RDD，其中每个元素等于前面输入元素的总和？

Question

我有一个 RDD 集合中的数字列表。从这个列表中，我需要创建另一个 RDD 列表，其中每个元素等于它之前所有元素的总和。如何在 Spark 中构建这样的 RDD？

以下 Scala 代码说明了我试图在 Spark 中实现的目标：

object Test {

  def main(args: Array[String]) {

    val lst: List[Float] = List(1, 2, 3)
    val result = sum(List(), 0, lst)
    println(result)
  }

  def sum(acc: List[Float], runningSum: Float, list: List[Float]): List[Float] = {
    list match {
      case List() => acc.reverse
      case List(x, _*) => {
        val newSum = runningSum + x
        sum(newSum :: acc, newSum, list.tail)
      }

    }
  }

运行此结果：

List(1.0, 3.0, 6.0)

此示例的等效 Spark 代码是什么？

Answer 1

正如其他人提到的那样，您的问题是 RDD 没有排序的概念，因此您必须提供“所有在它之前的元素”的定义作为输入的一部分，例如，作为行：

1 1.0 2 3.0 3 7.0

当然，我假设您无法将列表加载到驱动程序的内存中。鉴于您的问题类型，恐怕没有简单的解决方案，但我会尽力概述一个。

您的问题基本上需要顺序读取，但您可以分小块进行。例如，您可以通过调用 map(x => (x._1%10000, x._1, x._2)) 将列表划分为 10000 个项目的块。您可以根据内存限制调整的实际数字。

然后，您可以使用groupByKey() 方法将给定子列表中的所有项目放在一起。这会将整个子列表放在内存中的给定执行程序上。

然后，您可以像在示例中使用 sum() 方法一样对生成的子列表进行排序和计算。这会给你一个包含桶顺序、总和和列表的元组：

(1,3270,List(17, 12, 15)

您可以将此 RDD 映射到另外两个 RDD，首先，仅映射到 (1,3271) 以获取子列表的总和。其次，映射到 (1, 17), (1,12)... 元组，可以持久化。

您可以收集的第一张地图（它非常小，每 10000 个值只有 1 个元组）。一旦在驱动程序上，排序和求和以获得所有先前子列表的总和。把它作为一个新的 RDD 放回去，然后你可以用上面的第二个 RDD 做一个join。然后你可以将桶的总和（全局）添加到子列表（本地）的总和中，然后你可以将所有结果存储回磁盘。

总的来说，我所描述的是一种分而治之的方法。基本上将您的列表转到列表列表中，其中每个较小的列表都适合内存。

Answer 2

所以这在 Spark 中有点棘手，但一种选择是执行 mapPartitionsWithIndex 并计算每个分区的总和，然后将结果收集回驱动程序。然后，您可以使用它并执行另一个 mapPartitionsWithIndex 并使用先前分区的总和作为起点来计算总和，然后从这一点开始继续添加元素。这会有点慢，但我们不需要依赖收集或将所有元素带回驱动程序。

例如

val c = rdd.mapPartitionsWithIndex((f, i) => List((f, i.sum)).iterator).collect()
val sumRdd = rdd.mapPartitionsWithIndex{(f, i) => 
  val previousElem = c.filter(_._1 < f).map(_._2).sum
  var e = previousElem
  i.map{ ne =>
    val ret = e
    e = e + ne
    ret
   }}