clojure.java.jdbc /查詢大型結果集懶惰

Question

我想從數據庫讀取數百萬行並寫入文本文件。

這是我的問題的延續，現在database dump to text file with side effects

我的問題似乎是，記錄不會發生，直到程序完成。另一個我沒有處理的指標是，直到程序結束，文本文件才被寫入。

根據IRC的提示，似乎我的問題可能與:result-set-fn有關，並且在代碼的clojure.java.jdbc/query區域中默認爲doall。

我試圖用for函數替換它，但仍然發現內存消耗很高，因爲它將整個結果集拉入內存。

我怎樣纔能有一個:result-set-fn不拉像doall像一切？如何在程序運行時逐漸寫入日誌文件，而不是在執行完成後轉儲所有內容？

(let [ 
      db-spec    local-postgres 
      sql     "select * from public.f_5500_sf " 
      log-report-interval 1000 
      fetch-size   100 
      field-delim   "\t"                 
      row-delim   "\n"                 
      db-connection  (doto (j/get-connection db-spec) (.setAutoCommit false)) 
      statement   (j/prepare-statement db-connection sql :fetch-size fetch-size) 
      joiner    (fn [v] (str (join field-delim v) row-delim))      
      start    (System/currentTimeMillis)            
      rate-calc   (fn [r] (float (/ r (/ (- (System/currentTimeMillis) start) 100)))) 
      row-count   (atom 0)                
      result-set-fn  (fn [rs] (lazy-seq rs)) 
      lazy-results   (rest (j/query db-connection [statement] :as-arrays? true :row-fn joiner :result-set-fn result-set-fn)) 
      ]; }}} 
     (.setAutoCommit db-connection false) 
     (info "Started dbdump session...")  
     (with-open [^java.io.Writer wrtr (io/writer "output.txt")] 
     (info "Running query...")  
     (doseq [row lazy-results] 
      (.write wrtr row) 
     )) 
     (info (format "Completed write with %d rows" @row-count)) 
    ) 

Answer 1

我把最近修復clojure.java.jdbc通過將[org.clojure/java.jdbc "0.3.0-beta1"]我project.clj依賴上市。這一個增強/糾正了描述爲here的clojure.java.jdbc/query的:as-arrays? true功能。

我認爲這有所幫助，但我仍然可以覆蓋:result-set-fn到vec。

核心問題已通過將所有行邏輯打包爲:row-fn來解決。最初的OutOfMemory問題必須通過遍歷j/query結果集而不是定義具體的:row-fn來完成。

新的（工作），代碼如下：

(defn -main [] 
    (let [; {{{ 
     db-spec    local-postgres 
     source-sql   "select * from public.f_5500 " 
     log-report-interval 1000 
     fetch-size   1000 
     row-count   (atom 0) 
     field-delim   "\u0001" ; unlikely to be in source feed, 
             ; although i should still check in 
             ; replace-newline below (for when "\t" 
             ; is used especially) 
     row-delim   "\n" ; unless fixed-width, target doesn't 
            ; support non-printable chars for recDelim like 
     db-connection  (doto (j/get-connection db-spec) (.setAutoCommit false)) 
     statement   (j/prepare-statement db-connection source-sql :fetch-size fetch-size :concurrency :read-only) 
     start    (System/currentTimeMillis) 
     rate-calc   (fn [r] (float (/ r (/ (- (System/currentTimeMillis) start) 100)))) 
     replace-newline  (fn [s] (if (string? s) (clojure.string/replace s #"\n" " ") s)) 
     row-fn    (fn [v] 
           (swap! row-count inc) 
           (when (zero? (mod @row-count log-report-interval)) 
           (info (format "wrote %d rows" @row-count)) 
           (info (format "\trows/s %.2f" (rate-calc @row-count))) 
           (info (format "\tPercent Mem used %s " (memory-percent-used)))) 
           (str (join field-delim (doall (map #(replace-newline %) v))) row-delim)) 
     ]; }}} 
    (info "Started database table dump session...") 
    (with-open [^java.io.Writer wrtr (io/writer "./sql/output.txt")] 
     (j/query db-connection [statement] :as-arrays? true :row-fn 
       #(.write wrtr (row-fn %)))) 
    (info (format "\t\t\tCompleted with %d rows" @row-count)) 
    (info (format "\t\t\tCompleted in %s seconds" (float (/ (- (System/currentTimeMillis) start) 1000)))) 
    (info (format "\t\t\tAverage rows/s %.2f" (rate-calc @row-count))) 
    nil) 
) 

其他的事情我嘗試（有限的成功）所涉及的音色記錄並關閉了參考標準;我想知道如果使用REPL它可能會在顯示回我的編輯器（vim壁爐）之前緩存結果，我不確定這是否利用了大量內存。

另外，我用(.freeMemory (java.lang.Runtime/getRuntime))免費在記憶體周圍添加了記錄部件。我對VisualVM並不熟悉，並且確切地指出了我的問題所在。

我很滿意現在的工作方式，感謝大家的幫助。

Answer 2

您可以使用prepare-statement和:fetch-size選項。否則，儘管結果以惰性序列傳遞，但查詢本身仍然非常渴望。

prepare-statement需要一個連接對象，所以你需要明確地創建一個連接對象。這裏有您的使用情況會如何看一個例子：

(let [db-spec local-postgres 
     sql  "select * from big_table limit 500000 " 
     fetch-size 10000 ;; or whatever's appropriate 
     cnxn  (doto (j/get-connection db-spec) 
        (.setAutoCommit false)) 
     stmt  (j/prepare-statement cnxn sql :fetch-size fetch-size) 
     results (rest (j/query cnxn [stmt]))] 
    ;; ... 
) 

另一個選項

因爲這個問題似乎與query，嘗試with-query-results。它被認爲已被棄用，但仍然存在並且可行。下面是一個例子用法：

(let [db-spec local-postgres 
     sql  "select * from big_table limit 500000 " 
     fetch-size 100 ;; or whatever's appropriate 
     cnxn  (doto (j/get-connection db-spec) 
        (.setAutoCommit false)) 
     stmt  (j/prepare-statement cnxn sql :fetch-size fetch-size)] 
    (j/with-query-results results [stmt] ;; binds the results to `results` 
    (doseq [row results] 
     ;; 
    ))) 

Answer 3

我已經找到了一個更好的解決方案：您需要在事務中聲明一個遊標並從中獲取數據塊。例如：

(db/with-tx 
    (db/execute! "declare cur cursor for select * from huge_table") 
    (loop [] 
     (when-let [rows (-> "fetch 10 from cur" db/query not-empty)] 
     (doseq [row rows] 
      (process-a-row row)) 
     (recur)))) 

這裏，db/with-tx，db/execute!和db/query在db命名空間中聲明自己的快捷鍵：

(def ^:dynamic 
    *db* {:dbtype "postgresql" 
     :connection-uri <some db url>)}) 

(defn query [& args] 
    (apply jdbc/query *db* args)) 

(defn execute! [& args] 
    (apply jdbc/execute! *db* args)) 

(defmacro with-tx 
    "Runs a series of queries into transaction." 
    [& body] 
    `(jdbc/with-db-transaction [tx# *db*] 
    (binding [*db* tx#] 
     ~@body)))