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)))