6

有没有一种简单的方法来使用 scala 并行集合而不将完整集合加载到内存中？

例如，我有一个大集合，我想仅在一个适合内存的小块上并行执行特定操作（折叠），而不是在另一个块上等等，最后重新组合所有块的结果。

我知道，可以使用演员，但使用 par-collections 真的很好。

我已经写了一个解决方案，但它并不好：

  def split[A](list: Iterable[A], chunkSize: Int): Iterable[Iterable[A]] = {
    new Iterator[Iterable[A]] {
      var rest = list
      def hasNext = !rest.isEmpty
      def next = {
        val chunk = rest.take(chunkSize)
        rest = rest.drop(chunkSize)
        chunk
      }
    }.toIterable
  }                                               

  def foldPar[A](acc: A)(list: Iterable[A], chunkSize: Int, combine: ((A, A) => A)): A = {
    val chunks: Iterable[Iterable[A]] = split(list, chunkSize)
    def combineChunk: ((A,Iterable[A]) => A) = { case (res, entries) => entries.par.fold(res)(combine) }
    chunks.foldLeft(acc)(combineChunk)
  }                                               

  val chunkSize = 10000000                        
    val x = 1 to chunkSize*10                 

    def sum: ((Int,Int) => Int) = {case (acc,n) => acc + n }

    foldPar(0)(x,chunkSize,sum)
4

1 回答 1

4

您的想法非常简洁，可惜还没有这样的功能可用（AFAIK）。

我只是将您的想法改写为更短的代码。首先，我觉得对于并行折叠，使用幺半群的概念很有用——它是一种具有关联操作和零元素的结构。关联性很重要，因为我们不知道合并并行计算的结果的顺序。零元素很重要，因此我们可以将计算分成块并从零开始折叠每个块。不过，这并没有什么新意，这正是foldScala 的集合所期望的。

// The function defined by Monoid's apply must be associative
// and zero its identity element.
trait Monoid[A]
  extends Function2[A,A,A]
{
  val zero: A
}

接下来，Scala 的Iterators 已经有了一个有用的方法grouped(Int): GroupedIterator[Seq[A]]，可以将迭代器分割成固定大小的序列。和你的很相似split。这允许我们将输入切割成固定大小的块，然后对它们应用 Scala 的并行收集方法：

def parFold[A](c: Iterator[A], blockSize: Int)(implicit monoid: Monoid[A]): A =
  c.grouped(blockSize).map(_.par.fold(monoid.zero)(monoid))
                      .fold(monoid.zero)(monoid);

我们使用并行集合框架折叠每个块，然后（没有任何并行化）组合中间结果。

一个例子：

// Example:
object SumMonoid extends Monoid[Long] {
  override val zero: Long = 0;
  override def apply(x: Long, y: Long) = x + y;
}
val it = Iterator.range(1, 10000001).map(_.toLong)
println(parFold(it, 100000)(SumMonoid));
于 2013-07-04T19:02:36.713 回答