需要将大型 QueryRunner 结果流式传输到文件，似乎存储在内存中

Question

我正在尝试构建一个 Java 应用程序，该应用程序可以将任意 SQL SELECT 查询的非常大的结果集流式传输到 JSONL 文件中，特别是通过 SQLServer 但希望与任何 JDBC 一起运行DataSource。在 Python 中，这很容易将 sql 客户端结果视为生成器，然后调用json.dumps(). 但是，在这段代码中，它似乎在写出之前将所有内容都放入了内存中，这通常会导致堆和垃圾收集异常。我需要运行它的查询非常大，最多可以带回 10GB 的原始数据。执行时间不是主要问题，只要它每次都有效。我试过在每一行之后调用flush（这很荒谬），这似乎对小数据集有帮助，但对大数据集没有帮助。任何人都可以提出一个我可以用来轻松实现这一目标的策略吗？在我的 SQL 客户端类中，我使用 Apache DbUtilsQueryRunner并MapListHandler创建一个Maps 列表，这是我需要的灵活性（与 Java 中需要指定模式和类型的更传统方法相比）：public List<Map<String, Object>> query(String queryText) {    try {        DbUtils.loadDriver("com.microsoft.sqlserver.jdbc.Driver");        // this function just sets up all the connection properties. Ommitted for clarity        DataSource ds = this.initDataSource();        StatementConfiguration sc = new StatementConfiguration.Builder().fetchSize(10000).build();        QueryRunner queryRunner = new QueryRunner(ds, sc);        MapListHandler handler = new MapListHandler();        return queryRunner.query(queryText, handler);    } catch (Exception e) {        logger.error(e.getMessage());        e.printStackTrace();        return null;    }}JsonLOutputWriter班级：JsonLOutputWriter(String filename) {    GsonBuilder gsonBuilder = new GsonBuilder();    gsonBuilder.serializeNulls();    this.gson = gsonBuilder.create();    try {        this.writer = new PrintWriter(new File(filename), ENCODING);    } catch (FileNotFoundException | UnsupportedEncodingException e) {        e.printStackTrace();    }}void writeRow(Map row) {    this.writer.println(this.gson.toJson(row));}void flush() {    this.writer.flush();}主要方法：JsonLOutputWriter writer = new JsonLOutputWriter(outputFile)for (Map row : client.query(inputSql)) {    writer.writeRow(row);}writer.flush()

Answer 1

基本上这不能在DbUtils开箱即用的情况下完成。我摆脱了QueryRunner并且MapListHandler因为处理程序创建了一个ArrayList. 我不是基于拉，而是基于推，创建了一个非常相似的方法MyQueryRunner，它需要 aMyRowHandler而不是返回一个集合，而是迭代ResultSet并调用我的输出函数。

我确信有更优雅的方法可以做到这一点并返回某种行缓冲区，但这是我需要的 80/20 并且适用于大型数据集。

行处理程序

public class RowHandler {

    private static final RowProcessor ROW_PROCESSOR = new BasicRowProcessor();

    private JsonLOutputWriter writer;

    public RowHandler(JsonLOutputWriter writer) {

        this.writer = writer;

    }

    int handle(ResultSet rs) throws SQLException {

        AtomicInteger counter = new AtomicInteger();

        while (rs.next()) {

            writer.writeRow(this.handleRow(rs));

            counter.getAndIncrement();

        }

        return counter.intValue();

    }

    protected Map<String, Object> handleRow(ResultSet rs) throws SQLException {

        return this.ROW_PROCESSOR.toMap(rs);

    }

}

查询处理程序

class CustomQueryRunner extends AbstractQueryRunner {

    private final RowHandler rh;

    CustomQueryRunner(DataSource ds, StatementConfiguration stmtConfig, RowHandler rh) {

        super(ds, stmtConfig);

        this.rh = rh;

    }

    int query(String sql) throws SQLException {

        Connection conn = this.prepareConnection();

        return this.query(conn, true, sql);

    }

    private int query(Connection conn, boolean closeConn, String sql, Object... params)

            throws SQLException {

        if (conn == null) {

            throw new SQLException("Null connection");

        }

        PreparedStatement stmt = null;

        ResultSet rs = null;

        int count = 0;

        try {

            stmt = this.prepareStatement(conn, sql);

            this.fillStatement(stmt, params);

            rs = this.wrap(stmt.executeQuery());

            count = rh.handle(rs);

        } catch (SQLException e) {

            this.rethrow(e, sql, params);

        } finally {

            try {

                close(rs);

            } finally {

                close(stmt);

                if (closeConn) {

                    close(conn);

                }

            }

        }

        return count;

    }

}