在文件处理任务中使用适当的功能样式

Question

我有一个输入 csv 文件，需要为每个输入行生成一个输出文件。每个输入行可以是特定类型（比如“旧”或“新”），只能通过处理输入行来确定。

除了生成输出文件之外，我们还想打印输入文件中每种类型的行数的摘要。我的实际任务涉及根据输入行类型生成不同的 SQL，但为了保持示例代码的重点，我保持函数 proc-line 中的处理简单。函数 func 确定输入行的类型——再一次，我通过随机生成一个类型来保持它的简单性。实际逻辑更复杂。

我有以下代码，它可以完成这项工作。但是，为了保留生成摘要任务的功能样式，我选择返回一个关键字来表示每一行的类型，并创建了一个惰性序列来生成最终摘要。在命令式风格中，我们只需为每种线型增加一个计数。仅仅为了汇总而生成一个可能很大的集合似乎效率低下。我编码方式的另一个后果是重复 (.write writer ...) 部分。理想情况下，我只会编写一次代码。

对于消除我发现的两个问题（以及其他问题）有何建议？

(ns file-proc.core
  (:gen-class)
  (:require [clojure.data.csv :as csv]
            [clojure.java.io :as io]))

(defn func [x]
  (rand-nth [true false]))

(defn proc-line [line writer]
  (if (func line)
    (do (.write writer (str line "\n")) :new)
    (do (.write writer (str (reverse line) "\n")) :old)))

(defn generate-report [from to]
  (with-open
     [reader (io/reader from)
      writer (io/writer to)]
     (->> (csv/read-csv reader)
          (rest)
          (map #(proc-line % writer))
          (frequencies)
          (doall))))

Answer 1

我会尝试将数据处理与读/写文件等副作用分开。希望这将使 IO 操作保持在管道的相反边界，并且“中间”处理逻辑不知道输入来自哪里以及输出去向。

(defn rand-bool [] (rand-nth [true false]))
(defn proc-line [line]
  (if (rand-bool)
    [line :new]
    [(reverse line) :old]))

proc-line 不再需要写入器，它只关心 line 并返回已处理行的向量/2 元组以及关键字。它也不关心字符串格式——我们应该让csv/write-csv 来做。现在你可以这样做：

(defn process-lines [reader]
  (->> (csv/read-csv reader)
       (rest)
       (map proc-line)))

(defn generate-report [from to]
  (with-open [reader (io/reader from)
              writer (io/writer to)]
    (let [lines (process-lines reader)]
      (csv/write-csv writer (map first lines))
      (frequencies (map second lines)))))

这将起作用，但它将实现/将整个输入序列保留在内存中，而对于大文件，您不希望这样做。我们需要一种方法来保持这个管道的惰性/高效，但我们还必须在一次传递中从一个 和 产生两个“流”：处理后的行 only 是发送到write-csv，以及用于计算频率的每一行的元数据。一种“简单”的方法是引入一些可变性来跟踪元数据频率，因为 write-csv 消耗了惰性序列：

(defn generate-report [from to]
  (with-open [reader (io/reader from)
              writer (io/writer to)]
    (let [freqs (atom {})]
      (->> (csv/read-csv reader)
           ;; processing starts
           (rest)
           (map (fn [line]
                  (let [[row tag] (proc-line line)]
                    (swap! freqs update tag (fnil inc 0))
                    row)))
           ;; processing ends
           (csv/write-csv writer))
      @freqs)))

我删除了 process-lines 调用以使整个管道更加明显。当write-csv 完全（并且懒惰地）消耗了它的有效载荷时，freqs 将是一个类似{:old 23, :new 31} 的映射，这将是generate-report 的返回值。有改进/概括的空间，但我认为这是一个开始。

Answer 2

正如其他人所提到的，将写作和处理工作分开是理想的。以下是我通常这样做的方式：

(defn product-type [p]
  (rand-nth [:new :old]))

(defn row->product [row]
  (let [p (zipmap [:id :name :price] row)]
    (assoc p :type (product-type p))))

(defmulti to-csv :type)
(defmethod to-csv :new [product] ...)
(defmethod to-csv :old [product] ...)

(defn generate-report [from to]
  (with-open [rdr (io/reader from)
              wrtr (io/writer to)]
    (->> (rest (csv/read-csv rdr))
         (map row->product)
         (map #(do (.write wrtr (to-csv %)) %))
         (map :type)
         (frequencies)
         (doall))))

（代码可能无法运行——没有运行，抱歉。）

当然，构建哈希映射和使用多方法是可选的，但最好先为产品指定其类型。这样，它的数据决定了管道在做什么，而不是proc-line。

Answer 3

要重构代码，我们需要对generate-report 进行至少一项特征测试的安全网。由于该函数执行文件 I/O（稍后我们将使代码独立于 I/O），我们将使用此示例 CSV 文件，f1.csv：

Year,Code
1997,A
2000,B
2010,C
1996,D
2001,E

我们还不能编写测试，因为函数 func 使用 RNG，所以我们通过实际查看输入将其重写为确定性。在那里，我们将其重命名为new?，这更能代表问题：

(defn new? [row]
  (>= (Integer/parseInt (first row)) 2000))

为了练习，我们假设如果 Year 列 >= 2000，则该行是“新的”。

我们现在可以编写测试并查看它是否通过（为了简洁起见，我们只关注频率计算，而不是输出转换）：

(deftest characterization-as-posted
  (is (= {:old 2, :new 3}
         (generate-report "f1.csv" "f1.tmp"))))

现在进行重构。主要思想是意识到我们需要一个累加器，用reduce 替换map 并去掉frequencies 和doall。此外，我们将“line”重命名为“row”，因为这是在 CSV 格式中调用行的方式：

(defn generate-report [from to]                         ; 1
  (let [[old new _]                                     ; 2
        (with-open [reader (io/reader from)             ; 3
                    writer (io/writer to)]              ; 4
          (->> (csv/read-csv reader)                    ; 5
               (rest)                                   ; 6
               (reduce process-row [0 0 writer])))]     ; 7
    {:old old :new new}))                               ; 8

新的process-row（原process-line）变为：

(defn process-row [[old new writer] row]
  (if (new? row)
    (do (.write writer (str row "\n")) [old (inc new) writer])
    (do (.write writer (str (reverse row) "\n")) [(inc old) new writer])))

函数process-row，与任何要传递给reduce 的函数一样，有两个参数：第一个参数[old new writer] 是两个累加器和I/O 写入器的向量（向量已解构）；第二个参数row 是正在减少的集合中的一个元素。它返回新的累加器向量，该向量在集合的末尾在 generate-report 的第 2 行被解构，并在第 8 行用于创建一个与之前由 frequencies 返回的哈希映射等效的哈希映射。

我们可以做最后一次重构：将文件 I/O 与业务逻辑分开，这样我们就可以在没有准备好的输入文件脚手架的情况下编写测试，如下所示。

函数process-row变为：

(defn process-row [[old-cnt new-cnt writer] row]
  (let [[out-row old new] (process-row-pure old-cnt new-cnt row)]
    (do (.write writer out-row)
        [old new writer])))

并且业务逻辑可以通过纯（且易于测试）的功能来完成：

(defn process-row-pure [old new row]
  (if (new? row)
    [(str row "\n") old (inc new)]
    [(str (reverse row) "\n") (inc old) new]))

这一切都没有改变任何东西。

Answer 4

恕我直言，我会将两个不同的方面分开：计算频率和写入文件：

(defn count-lines
   ([lines] (count-lines lines 0 0))
   ([lines count-old count-new]
     (if-let [line (first lines)]
        (if (func line)
           (recur count-old (inc count-new) (rest lines))
           (recur (inc count-old) count-new (rest lines)))
        {:new count-new :old count-old})))


 (defn generate-report [from to]
   (with-open [reader (io/reader from)
               writer (io/writer to)]
     (let [lines (rest (csv/read-csv reader))
           frequencies (count-lines lines)]
         (doseq [line lines]
            (.write writer (str line "\n"))))))