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

Question

我正在尝试构建一个Java应用程序，该应用程序可以将任意SQL SELECT查询的非常大的结果集流式传输到JSONL文件中，特别是通过SQLServer，但我希望与任何JDBCDataSource一起运行。在Python中，只需将sql客户端结果视为生成器，然后调用json.dumps()就很容易了。然而，在这段代码中，它似乎在写出来之前将所有内容都放在内存中，这通常会导致堆和垃圾收集异常。我需要运行的查询非常大，可以返回高达10GB的原始数据。执行时间不是首要问题，只要它每次都有效。

我试过在每次行之后调用flush(这很荒谬)，这似乎对小数据集有帮助，但对大数据集没有帮助。有人能给我一个策略吗？

在我的SQL客户端类中，我使用Apache DbUtilsQueryRunner和MapListHandler来创建Map的列表，这是我所需要的灵活性(而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，它接受一个MyRowHandler，而不是返回一个集合，只是在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;
}
}