通过 Clojure 和 JDBC 将 5,000,000 行移动到另一个 Postgresql 数据库

Question

我正在尝试将 5,000,000 行从一个 Postgre DB 移动到另一个。两个连接都在 Hikari CP 连接池中。

我浏览了很多文档和帖子。它给我留下了下面的代码。但它并不是真的有用：

(jdbc/with-db-connection [tx {:datasource source-db}]
  (jdbc/query tx
      [(jdbc/prepare-statement (jdbc/get-connection tx)
                                answer-sql
                                {:fetch-size 100000})]
                  {:result-set-fn (fn [result-set]
                                    (jdbc/insert-multi!
                                     {:datasource target-db}
                                     :migrated_answers
                                     result-set))}))

我已经尝试了很多不同的形式。 jdbc/with-db-transaction 或任何其他我能想到的都没有多大帮助。

1. 很多教程和帖子只提到如何将结果作为一个整体进行处理。进入 RAM 的小桌子绝对没问题，但看起来很快。但事实并非如此。

2. 因此，当我正确使用 :fetch-size 并且我的 RAM 不会爆炸（诡计）时，传输速度非常慢，因为两个连接在 DB 端的“活动”和“事务空闲”状态之间切换。我从来没有等过这么久才能找到实际传输的任何数据！

   当我在 Talend Open Studio（生成 Java 代码的 ETL 工具）中创建这个简单的批处理时，它会在 5 分钟内传输所有数据。并且光标大小“也”设置为 100000 那里。我认为 Clojure 的干净代码应该更快。

3. 我得到的最快结果是使用下面的代码。我认为这是因为:as-array 参数。如果我不使用:max-rows参数内存会爆炸，因为它没有被延迟处理，所以我不能将它用于整个transfet。为什么？我不明白这里的规则。

   (jdbc/with-db-transaction [tx {:datasource source-db}]
     (jdbc/query tx
                 [(jdbc/prepare-statement (:connection tx)
                                           answer-sql
                                          {:result-type :forward-only
                                           :concurrency :read-only
                                           :fetch-size 2000
                                           :max-size 250000})]
                 {:as-arrays? true
                  :result-set-fn (fn [result-set]
                                   (let [keys (first result-set)
                                         values (rest result-set)]
                                     (jdbc/insert-multi! 
                                        {:datasource dct-db}
                                         :dim_answers
                                          keys values)))}))
   

如果我明显缺少任何帮助或信息，我将不胜感激。

Answer 1

我认为这里的关键观察结果是，虽然您的 query 是从一个 DB 懒惰地流式传输结果，但您的 insert 只是对另一个 DB 的巨大写入。关于内存使用情况，我认为如果您在最后收集所有这些结果（在内存中）以进行单个写入操作，那么您是否懒惰地流式传输查询结果并没有太大区别。

平衡内存使用与吞吐量的一种方法是批量写入：

(db/with-db-transaction [tx {:datasource source-db}]
  (db/query tx
    [(db/prepare-statement (:connection tx)
                           answer-sql
                           {:result-type :forward-only
                            :concurrency :read-only
                            :fetch-size 2000})]
    {:as-arrays? true
     :result-set-fn (fn [result-set]
                      (let [keys (first result-set)
                            values (rest result-set)]
                        (doseq [batch (partition-all 2000 values)]
                          (db/insert-multi! {:datasource dct-db}
                                            :dim_answers
                                            keys
                                            batch))))}))

不同之处在于它使用partition-all 批量插入values（与:fetch-size 大小相同，但我确信这可以调整）。通过将 JVM 最大堆大小设置为 -Xmx1g 之类的值，将这种方法的性能/内存使用情况与其他方法进行比较。我无法使用此堆大小完成非批处理版本。

我能够在大约 1 分钟内在笔记本电脑上的本地 PostgreSQL 数据库之间迁移 600 万条小型行，java 使用

如果您确实分批插入，如果适合您的用例，您可能需要考虑将所有插入包装在单个事务中。为简洁起见，我将附加交易留在这里。

如果我不使用:max-size参数内存爆炸

我在最新的 clojure.java.jdbc 中找不到此选项的任何参考（除了规范），它不影响我的测试。我确实看到了:max-rows，但你肯定不希望这样。

我认为是因为:as-array 参数。

我希望这对内存使用有益；行向量应该比行映射更节省空间。

Answer 2

这个解决方案最适合我，而且似乎比 Taylor 的解决方案更快。但是非常感谢你帮助我。

在事务完成之前它不会提交。我必须遇到任何问题，看看我是否不必拉皮条，但我现在很高兴。我尝试用with-db-connection 替换第一个事务，但它使记录直接加载到 RAM 中。

(defn data->transfer2 [sql table]
     (jdbc/with-db-transaction [read-tx {:datasource dag-db}]
     (jdbc/with-db-transaction [tx {:datasource dct-db}]
        (jdbc/query read-tx
                  [(jdbc/prepare-statement (:connection read-tx)
                                           answer-sql
                                           {:result-type :forward-only
                                            :concurrency :read-only
                                            :fetch-size 100000})]
                  {:as-arrays? true
                   :result-set-fn (fn [result-set]
                                    (let [keys (first result-set)
                                          values (rest result-set)]
                                      (doseq [btch (partition-all 100000 values)]
                                        (jdbc/insert-multi! tx
                                                            :dim_answers
                                                             keys
                                                             btch))))})))