scala - Iterator[Something] 到 Iterator[Seq[Something]]

Question

我需要处理一个“大”文件(内存中放不下的文件)。

我想批处理数据。比方说我想将它们插入到数据库中。但是由于它太大而无法放入内存，因此一个一个地处理元素也太慢了。

所以我想从 Iterator[Something] 到 Iterator[Iterable[Something]] 到批处理元素。

从这里开始:

CSVReader.open(new File("big_file"))
  .iteratorWithHeaders
  .map(Something.parse)
  .foreach(Jdbi.insertSomething)

我可以在 foreach 语句中使用可变序列做一些肮脏的事情并刷新每个 x 元素，但我确信有更聪明的方法来做到这一点......

// Yuk... :-(
val buffer = ArrayBuffer[Something]()
CSVReader.open(new File("big_file"))
  .iteratorWithHeaders
  .map(Something.parse)
  .foreach {
     something =>
       buffer.append(something)
       if (buffer.size == 1000) {
         Jdbi.insertSomethings(buffer.toList)
         buffer.clear()
       }
   }
Jdbi.insertSomethings(buffer.toList)

Answer 1

如果您的批处理可以具有固定大小(如您的示例所示)，则 Scala 的 Iterator 上的 grouped 方法完全符合您的要求:

val iterator = Iterator.continually(1)

iterator.grouped(10000).foreach(xs => println(xs.size))

这将在固定数量的内存中运行(当然不计算终端在内存中存储的任何文本)。

我不确定您的 iteratorWithHeaders 返回什么，但如果它是 Java 迭代器，您可以像这样将它转换为 Scala 迭代器:

import scala.collection.JavaConverters.

val myScalaIterator: Iterator[Int] = myJavaIterator.asScala

这将保持适当的惰性。