java - 使用 JooQ 从 CSV 中提取 "batch insert"并同时跟踪插入的记录？

Question

我有一个 CSV，它有 3400 万行长。是的，不是开玩笑。

这是由 parser tracer 生成的 CSV 文件然后导入到 corresponding debugging program .

而问题出在后者。

现在我一一导入所有行:

private void insertNodes(final DSLContext jooq)
    throws IOException
{
    try (
        final Stream<String> lines = Files.lines(nodesPath, UTF8);
    ) {
        lines.map(csvToNode)
            .peek(ignored -> status.incrementProcessedNodes())
            .forEach(r -> jooq.insertInto(NODES).set(r).execute());
    }
}

csvToNode 只是一个映射器，它将把 String(CSV 的一行)转换为 NodesRecord 以供插入。

现在，行:

            .peek(ignored -> status.incrementProcessedNodes())

好吧...方法名称几乎说明了一切；它会增加 status 中的一个计数器，该计数器反射(reflect)到目前为止已处理的行数。

发生的事情是每秒查询一次此 status 对象以获取有关加载过程状态的信息(我们在这里讨论的是 3400 万行；加载它们大约需要 15 分钟)。

但是现在 jooq 有这个(取自他们的文档)可以直接从 CSV 加载:

create.loadInto(AUTHOR)
      .loadCSV(inputstream)
      .fields(ID, AUTHOR_ID, TITLE)
      .execute();

(尽管我个人永远不会使用 .loadCSV() 重载，因为它不考虑 CSV 编码)。

当然，JooQ 会设法将其转化为合适的构造，以便对于这个或那个数据库引擎，吞吐量最大化。

然而，问题是我丢失了从当前代码中获得的“按秒”信息...如果我用 select count(*) from the_victim_table 替换查询，那种失败的重点，更不用说这可能需要很长时间了。

那么，我如何获得“两全其美”呢？也就是说，有没有一种方法可以使用“优化的 CSV 加载”并在任何时间足够快地查询到目前为止已插入了多少行？

(注意:如果有关系，我目前使用 H2；PostgreSQL 版本也在计划中)

Answer 1

有很多方法可以对此进行优化。

自定义负载分区

在您身边优化查询执行的一种方法是将值集收集到:

• 批量语句(如 INSERT INTO t VALUES(1), (2), (3), (4))
• 批处理语句(如在 JDBC 批处理中)
• 提交段(在 N 条语句后提交)

... 而不是一个一个地执行它们。这也是 Loader API 的作用(见下文)。所有这些措施都可以大大提高加载速度。

这是您当前可以“收听”加载进度的唯一方式。

使用 jOOQ 3.6+ 进行负载分区

(尚未发布，但很快就会发布)

jOOQ在jOOQ 3.6中原生实现了以上三种分区措施

使用供应商特定的 CSV 加载机制

jOOQ 将始终需要通过 JDBC，因此可能不会为您提供最快选项。大多数数据库都有自己的加载 API，例如你提到的那些:

• H2:http://www.h2database.com/html/tutorial.html#csv
• PostgreSQL:http://www.postgresql.org/docs/current/static/sql-copy.html

这会更底层，但肯定比其他任何东西都快。

一般说明

What happens is that this status object is queried every second to get information about the status of the loading process (we are talking about 34 million rows here; they take about 15 minutes to load).

这是一个非常有趣的想法。将此注册为 Loader API 的功能请求:Using JooQ to "batch insert" from a CSV _and_ keep track of inserted records at the same time?

though personally I'd never use THAT .loadCSV() overload since it doesn't take the CSV encoding into account

我们已经为 jOOQ 3.6 修复了这个问题，感谢您的评论:https://github.com/jOOQ/jOOQ/issues/4141

And of course JooQ will manage to turn that into a suitable construct so that for this or that DB engine the throughput is maximized.

不，jOOQ 不对最大化吞吐量做出任何假设。这非常困难，并且取决于数据库供应商以外的许多其他因素，例如:

• table 上的约束
• 表上的索引
• 打开/关闭日志记录
• 等等

jOOQ 帮助您自己最大化吞吐量。例如，在 jOOQ 3.5+ 中，您可以:

• 设置提交率(例如，每 1000 行提交一次)以避免在启用日志记录的情况下插入较长的 UNDO/REDO 日志。这可以通过 commitXXX() 方法完成。

在 jOOQ 3.6+ 中，您还可以:

• 设置批量语句速率(例如，将 10 行合并到一个语句中)以显着加快执行速度。这可以通过 bulkXXX() 方法完成。
• 设置批处理语句速率(例如，将 10 个语句合并到一个 JDBC 批处理中)以显着加快执行速度 (see this blog post for details)。这可以通过 batchXXX() 方法完成。