使用 clojure-csv.core 解析一个巨大的 csv 文件答案

【问题标题】：using clojure-csv.core to parse a huge csv file使用 clojure-csv.core 解析一个巨大的 csv 文件
【发布时间】：2013-11-07 20:05:24
【问题描述】：

到目前为止我有：

(:require [clojure-csv.core :as csv])
(:require [clojure.java.io :as io]))

(def csv-file (.getFile  (clojure.java.io/resource "verbs.csv")))

(defn process-csv [file]
  (with-open  [rdr  (io/reader file)]
    (csv/parse-csv rdr)))

但我收到了java.io.IOException: Stream closed。我正在使用clojure-csv，它公开了两种方法，我使用的第一种方法是parse-csv，文档说：

Takes a CSV as a char sequence or string, and returns a lazy sequence of vectors of strings

我想我知道：with-open 是懒惰的，(csv/parse-csv rdr))) 中的 rdr 是 csv 文件的单行，对吗？

PS。我也想搜索文件，重复打开文件（即使它正在被懒惰地读取）并搜索整个内容是否很昂贵？？

【问题讨论】：

如下所示，with-open 不是惰性的，但 parse-csv 是。 parse-csv 将惰性返回输入 csv 中的每个值，作为惰性向量列表（每行一个向量）。

标签： csv clojure bigdata

【解决方案1】：

问题在于您的process-csv 函数并没有真正“处理”with-open 范围内的 CSV 数据，而是将其作为惰性序列返回。当执行退出with-open 范围时，流已经关闭。稍后尝试遍历惰性列表将引发异常。

除非您确信可以将 CSV 文件完全读取并解析到内存中，否则我建议不要遵循其他答案中的建议，即强制评估 with-open 范围内的惰性序列使用doall.

相反，如果您希望将资源分配和解除分配部分与“更可重用”的业务逻辑分开，您应该这样做：

(defn process-csv [rdr conn]
  (doseq [row (csv/parse-csv rdr) :where (wanted? row)]
    (save-to-custom-database-table conn row)))

(defn start [process-fn]
  (let [csv-file (.getFile  (clojure.java.io/resource "verbs.csv"))]
    (with-open [rdr (jio/reader csv-file)
                conn (database-connection "TEST")]
      (process-fn rdr conn))))

(start process-csv)

如您所见，process-csv 函数以“抽象”方式处理读取器和数据库资源，即不关心这些资源是Closeable 并且应该在使用后关闭的事实。相反，资源的最终确定/关闭在 start 函数中作为单独的事项处理。

我还建议您研究 Clojure 协议，看看它们在上述类似场景中抽象资源方面的作用。

【讨论】：

【解决方案2】：

我知道这已经回答了，但这里有一个与@noisesmith 类似的解决方案，它创建了一个显式的惰性序列，如果你到达输入的末尾，它会自动关闭。

如果您打算懒惰地处理整个文件，这意味着您不必自己显式管理句柄，否则您会遇到打开的句柄问题。

(defn lazy-read-csv
  [csv-file]
  (let [in-file (io/reader csv-file)
        csv-seq (csv/read-csv in-file)
        lazy (fn lazy [wrapped]
               (lazy-seq
                 (if-let [s (seq wrapped)]
                   (cons (first s) (lazy (rest s)))
                   (.close in-file))))]
    (lazy csv-seq)))

这是来自 Eric Rochester 的出色 Clojure Data Analysis Cookbook

【讨论】：

我建议不要这样做，因为它可能会导致资源不被回收。如果消费者没有遍历整个列表（例如，使用(take n coll) 仅使用其中的一部分），那么阅读器将永远不会正确关闭。
正如我在我的解决方案中所说：“如果你要懒惰地处理整个文件......否则你有打开的句柄问题”。当您知道要懒惰地处理整个列表并让它自行清理时，此解决方案明确。

【解决方案3】：

with-open 不是惰性的，但是如果你在 with-open 内部做一些惰性的事情，如果惰性动作没有被强制在 with-open 的范围内，你可能会遇到问题。需要做的是在退出with-open块之前强制所有惰性结果。

(defn process-csv [file]
  (with-open [rdr (io/reader file)]
    (doall (csv/parse-csv rdr))))

doall 函数旨在确保实现整个惰性序列。

由于您输入的大小，另一种可能性是安排自己关闭阅读器，然后将惰性用于其预期目的（仅在您需要时生成结果）。

(defn find-results
 [stream]
 (for [record stream
       :while (seq (first record))]
   record))

(def rdr (io/reader "verbs.csv"))
(def csv (csv/parse-csv rdr))

(def results (doall (find-results csv)))

(.close rdr)

【讨论】：

【解决方案4】：

当文件已经关闭时，看起来文件正试图在 with-open 表单之外延迟解析。

尝试这样的事情来验证，打印前 5 行解析：

(defn process-csv [file]
  (with-open  [rdr  (io/reader file)]
    (let [lines (csv/parse-csv rdr)]
         (doseq [l (take 5 lines)]
            (println l)))))

如果文件很大，我不认为多次打开文件会比在里面搜索内容昂贵。

如果您需要多次这样做，我会考虑建立某种搜索索引。

【讨论】：