Scala：可变大小批次的迭代器的功能替换

Question

我正在使用 Scala 从我们的列存储 Cassandra 中读取列。每列包含多个条目，n，其中 n 可以在 10 到 20 之间。我们读取一批条目，即一次 1000 个，并且必须从条目创建列；每个条目都有一个附加的 ID，我们可以使用它来分组。

目前，我们使用迭代器来遍历批次中的条目，并通过比较当前 ID 和先前 ID 来确定我们是否进入了新列，并且我们读取了许多批次，直到完成。我们需要在每个批次迭代结束时存储部分列，因为该列的其余部分将在下一个批次中。我在下面放了一些伪代码来演示我们目前使用的基本算法。

如何以函数的方式做到这一点？（如果 n 是常数，这将是一个简单的问题，因为我们可以适当地设置批量大小。）

伪代码：

val resultBuffer // collects all columns
val columnBuffer // collects entries for current column
var currentId    // id of current column

while(batchIterator.hasNext){
     val batch = batchIterator.getNext
     val entryIterator = batch.entries.iterator

     while(entryIterator.hasNext){
           val entry = entryIterator.next
            if(entry.id != currentId) {
               currentId = entry.id  
               resultBuffer += columnBuilder(columnBuffer)
               columnBuffer.removeAll
               columnBuffer += entry
            } else {
                columnBuffer += entry
            } 
     }
}

Answer 1

这是一个更实用的实现草图，它使用滑动来对条目迭代器中的条目进行分组：

val resultBuffer // collects all columns

batchIterator.foreach(batch => {
  val buffer = 
    batch.entries.sliding(2).foldLeft(new ColumnBuffer){(buffer, (curr, next)) =>
      if (curr.id != next.id) {
        resultBuffer += columnBuilder(buffer :+ entry /* Append entry to buffer */)
        new ColumnBuffer
      } else
        buffer += entry /* Return buffer with entry added */
    }

  if (buffer.nonEmpty) resultBuffer += columnBuilder(buffer)
}

这里，唯一的“全局”对象必须是可变的，是resultBuffer。我们甚至可以通过将其作为另一个累加器包含在内部 foldLeft 中并将外部 foreach 替换为另一个 foldLeft 来摆脱它。

如果运行时效率对您的代码至关重要，那么您绝对应该对各种可能的实现进行基准测试，以便在功能性和性能之间找到良好的折衷。

编辑 1： 修复了草图中的错误，即存储在 buffer 中的最后一个条目序列未添加到 resultBuffer。错误已存在于 OP 的代码中。

编辑 2：（针对 Chuck 的评论）

curr 将取值entries(0) 到entries(entries.size() - 2)，即不会处理最后一个元素。解决此问题的一种方法是将虚拟元素附加到迭代器，例如

(batch.entries ++ List(dummy)).sliding(2).foldLeft ...

这不好，更重要的是，当batch.entries 为空时，它不会工作，因为sliding(2) 然后会生成单个窗口List(dummy)。另一种解决方案是将next 包含在内部foldLeft 的累加器中，并在foldLeft 终止后处理它。我还没有解决，但看起来这会使解决方案更不吸引人。