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

Question

我试图从一个Postgre数据库移动5,000,000行到另一个。两个连接都在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没有爆炸（hocus pocus）而不是传输IS非常慢，两个连接在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

我认为这里的关键观察是，虽然您的查询是从一个数据库延迟流式传输结果，但您的插入只是对另一个数据库的一个巨大写入。关于内存使用情况，如果您在最后收集所有这些结果（内存中）以进行单个写入操作，我认为您是否懒得流式传输查询结果并不会有太大区别。

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

(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万条小行，并使用<400MB内存的java。我也用过HikariCP。

如果您批量插入，您可能需要考虑在单个事务中包装所有插入，如果它适合您的用例。为简洁起见，我在这里留下了额外的交易。

如果我不使用: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))))})))