我正在嘗試構建一個 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()
1 回答
一只甜甜圈
TA貢獻1836條經驗 獲得超5個贊
基本上這不能在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;
}
}
添加回答
舉報
0/150
提交
取消