java - 将表从源数据库复制到同一主机系统上的目标数据库 (java.lang.OutOfMemoryError)

Question

我需要查询一个数据库并将结果集复制到另一个具有相同数据库结构并且也在同一主机系统上的数据库。

如果查询结果非常小，以下 JAVA 函数运行良好(快速且无错误):

public void copyTableData(Connection dbConnOnSrcDB, Connection dbConnOnDestDB,
                          String sqlQueryOnSrcDB, String tableNameOnDestDB) 
  throws SQLException {

    try (

        PreparedStatement prepSqlStatmOnSrcDB = dbConnOnSrcDB.prepareStatement(sqlQueryOnSrcDB);

        ResultSet sqlResultsFromSrcDB = prepSqlStatmOnSrcDB.executeQuery()
    ) {
        ResultSetMetaData sqlMetaResults = sqlResultsFromSrcDB.getMetaData();

        // Stores the query results
        List<String> columnsOfQuery = new ArrayList<>();

        // Store query results
        for (int i = 1; i <= sqlMetaResults.getColumnCount(); i++)
            columnsOfQuery.add(sqlMetaResults.getColumnName(i));

        try (
            PreparedStatement prepSqlStatmOnDestDB = dbConnOnDestDB.prepareStatement(
                "INSERT INTO " + tableNameOnDestDB +
                     " (" + columnsOfQuery.stream().collect(Collectors.joining(", ")) + ") " +
                        "VALUES (" + columnsOfQuery.stream().map(c -> "?").collect(Collectors.joining(", ")) + ")")
        ) {

            while (sqlResultsFromSrcDB.next()) {
                for (int i = 1; i <= sqlMetaResults.getColumnCount(); i++)
                    prepSqlStatmOnDestDB.setObject(i, sqlResultsFromSrcDB.getObject(i));

                prepSqlStatmOnDestDB.addBatch();
            }
            prepSqlStatmOnDestDB.executeBatch();
        }
    }
}

但是我有数百兆字节范围内的非常大的数据库查询和结果集。

问题 A: 我发现在处理第二行代码时引发了以下 OutOfMemoryError:

ResultSet sqlResultsFromSrcDB = prepSqlStatmOnSrcDB.executeQuery()

JAVA-Exception:

Exception in thread "main" java.lang.OutOfMemoryError: Java heap space
at java.lang.Class.getDeclaredFields0(Native Method)
at java.lang.Class.privateGetDeclaredFields(Class.java:2583)
at java.lang.Class.getDeclaredField(Class.java:2068)
at java.util.concurrent.atomic.AtomicReferenceFieldUpdater$AtomicReferenceFieldUpdaterImpl$1.run(AtomicReferenceFieldUpdater.java:323)
at java.util.concurrent.atomic.AtomicReferenceFieldUpdater$AtomicReferenceFieldUpdaterImpl$1.run(AtomicReferenceFieldUpdater.java:321)
at java.security.AccessController.doPrivileged(Native Method)
at java.util.concurrent.atomic.AtomicReferenceFieldUpdater$AtomicReferenceFieldUpdaterImpl.<init>(AtomicReferenceFieldUpdater.java:320)
at java.util.concurrent.atomic.AtomicReferenceFieldUpdater.newUpdater(AtomicReferenceFieldUpdater.java:110)
at java.sql.SQLException.<clinit>(SQLException.java:372)
at org.postgresql.core.v3.QueryExecutorImpl.processResults(QueryExecutorImpl.java:2156)
at org.postgresql.core.v3.QueryExecutorImpl.execute(QueryExecutorImpl.java:300)
at org.postgresql.jdbc.PgStatement.executeInternal(PgStatement.java:428)
at org.postgresql.jdbc.PgStatement.execute(PgStatement.java:354)
at org.postgresql.jdbc.PgPreparedStatement.executeWithFlags(PgPreparedStatement.java:169)
at org.postgresql.jdbc.PgPreparedStatement.executeQuery(PgPreparedStatement.java:117)
at Application.copyTableData(Application.java:159)
at Application.main(Application.java:585)

问题 B: 复制工作确实需要很多时间。有没有办法加快复制过程？

数据库查询是:

String[] tables = new String[]{
                    "table1",
                    "table1_properties",
                    "table1_addresses",
                    "table2",
                    "table3",
                    "table4",
                    "table5",
                    "table6",
                    "table7",
                    "table8",
                    "table9",
                    "table10"
            };

函数调用:

for( String table : tables ){

  getDataFromSrcDB = "SELECT " + table + ".* " +
    "FROM table1 " +
        "FULL JOIN table1_properties " +
            "ON table1_properties.d_id=table1.d_id " +
        "FULL JOIN table1_addresses " +
            "ON table1_addresses.d_id=table1_properties.d_id " +
        "FULL JOIN table2 " +
            "ON table2.p_id=table1_properties.p_id " +
        "FULL JOIN table3 " +
            "ON table3.d_id=table1.d_id " +
        "FULL JOIN table4 " +
            "ON table4.d_id=table1.d_id " +
        "FULL JOIN table5 " +
            "ON table5.d_id=table1.d_id " +
        "FULL JOIN table6 " +
            "ON table6.d_id=table1.d_id " +
        "FULL JOIN table7 " +
            "ON table7.d_id=table1.d_id " +
        "FULL JOIN table8 " +
            "ON table8.id=table4.id " +
        "FULL JOIN table9 " +
            "ON table9.d_id=table1.d_id " +
        "FULL JOIN table10 " +
            "ON table10.a_id=table1_addresses.a_id " +
        "WHERE ST_Intersects(ST_MakeEnvelope(" +
               minLong + "," +
               minLat + "," +
               maxLong + "," +
               maxLat + ",4326), geom :: GEOMETRY) OR " +
        "ST_Intersects(ST_MakeEnvelope(" +
               minLong + "," +
               minLat + "," +
               maxLong + "," +
               maxLat + ",4326), CAST(table3.location AS GEOMETRY))";

   copyTableData(dbConnOnSrcDB, dbConnOnDestDB, getDataFromSrcDB, table);
}

Answer 1

当批处理的大小很大时，你会得到这个错误:

Exception in thread "main" java.lang.OutOfMemoryError: Java heap space

我有一些解决办法。

第一个解决方案

相反，您可以将批处理划分为小批处理，例如每个 1_000 保留数据，您还需要一些配置，如 Mark Rotteveel在评论中提及，并作为文档提及 Getting results based on a cursor :

By default the driver collects all the results for the query at once. This can be inconvenient for large data sets so the JDBC driver provides a means of basing a ResultSet on a database cursor and only fetching a small number of rows.

那么你应该做什么:

• 与服务器的连接必须使用 V3 协议(protocol)。
• 连接不能处于自动提交模式。
• 给定的查询必须是单个语句
• Statement的fetch size需要适当的大小
• ..阅读文档中的详细信息

在这种情况下，您的代码可以是这样的:

//Note here you set auto commit for the source connection
dbConnOnSrcDB.setAutoCommit(false);

final int batchSize = 1000;
final int fetchSize = 50;
int count = 0;
...
//Set the appropriate size for the FetchSize
sqlResultsFromSrcDB.setFetchSize(fetchSize);
while (sqlResultsFromSrcDB.next()) {
    for (int i = 1; i <= sqlMetaResults.getColumnCount(); i++) {
        prepSqlStatmOnDestDB.setObject(i, sqlResultsFromSrcDB.getObject(i));
    }
    prepSqlStatmOnDestDB.addBatch();
    if (++count % batchSize == 0) {
        prepSqlStatmOnDestDB.executeBatch();
    }
}
prepSqlStatmOnDestDB.executeBatch(); // insert remaining records

---

第二种解决方案

因为您使用的是 PostgreSQL，所以我想使用 dblink在数据库之间将数据传输到另一个数据库。

---

一些有用的链接:

• https://viralpatel.net/blogs/batch-insert-in-java-jdbc/
• How to use (install) dblink in PostgreSQL?
• http://www.postgresonline.com/journal/archives/44-Using-DbLink-to-access-other-PostgreSQL-Databases-and-Servers.html