scala - 在函数式 Scala 代码中使用 JDBC PreparedStatement.addBatch

Question

我需要从一个数据库读取数百万行并将它们写入另一个数据库。我想使用 PreparedStatement.addBatch 进行大批量写入(可能 1000 行)。我不需要他们参与交易。我正在用 Scala 2.9.2 编写代码。

一种方法如下:

val sourceResultSet = ...
val targetStatement = targetConnection.prepareStatement(...)
var rowCount = 0
while (sourceResultSet.next()) {
  // Read values from sourceResultSet and write them to targetStatement
  targetStatement.addBatch()
  rowCount += 1
  if (rowCount % 1000 == 0) {
    targetStatement.executeBatch()
    rowCount = 0
  }
}

如何在不使用 var rowCount 的情况下以更实用的方式执行此操作？我还需要考虑 RAM 的使用情况；我正在读取数百万行，因此任何涉及同时将所有源行存储在内存中的解决方案都会失败。

Answer 1

sourceResultSet 的类型是什么？我假设根据您的使用情况使用 Iterator/Stream，但无论哪种方式，您都可以使用 Scala 集合的 take 一次获取 1000 个元素(这适用于 Lists、Sets、Iterators、Streams 等) .为了更有效地做到这一点(虽然只有副作用，所以不是纯功能性的)，定义一个内联函数:

def processSource(sourceResultSet: Iterator): Unit = {
  if(sourceResultSet.hasNext) {
    sourceResultSet.take(1000).foreach(row => /* Add to batch */)
    targetStatement.executeBatch()
    processResult(sourceResultSet) // How you handle the recursion depends on what sourceResultSet is
  }
}

val sourceResultSet = ...
val targetStatement = targetConnection.prepareStatement(...)
processSource(sourceResultSet)

只要 sourceResultSet 是惰性的(Stream 或 Iterator)，这将避免立即将其加载到内存中。