为什么r/fold没有并行化到所有可用的 cpu 上？

Question

我本来期望下述代码能在所有可用的 cpu 上进行并行化（在这个案例中为 12 个 CPU），但它只在一个 CPU 上运行。我错在哪里了？

(ns csv2summap.core
  (:require [clojure.data.csv :as csv]
            [clojure.java.io :as io]
            [clojure.core.reducers :as r])))

(with-open [writer (io/writer "numbers.csv")]
  (csv/write-csv
   writer
   (take 10000000
         (repeatedly #(vector (char (+ 65 (rand-int 26))) (rand-int 1000))))))

(defn sum-vals
([] {})
([m [k v]]
 (update m k (fnil + 0) (Integer/parseInt v))))

(defn merge-sums
([] {})
([& m] (apply merge-with + m)))

(time
(with-open [reader (io/reader "numbers.csv")]
  (doall
   (r/fold
    (/ 10000000 12)
    merge-sums
    sum-vals
    (csv/read-csv reader)))))

(def n-cpu (.availableProcessors (Runtime/getRuntime)))
=>12

(ns csv2summap.core
  (:require [clojure.data.csv :as csv]
            [clojure.java.io :as io]
            [clojure.core.reducers :as r])))

-

(defproject csv2summap "0.1.0-SNAPSHOT"
  :description "FIXME: write description"
  :url "http://example.com/FIXME"
  :license {:name "EPL-2.0 OR GPL-2.0-or-later WITH Classpath-exception-2.0"
            :url "https://www.eclipse.org/legal/epl-2.0/"}
  :dependencies [[org.clojure/clojure "1.10.1"]
                 [org.clojure/data.csv "1.0.0"]]
  :repl-options {:init-ns csv2summap.core})

Answer 1

感谢Clojurians Slack频道中的Sean Corfield和Adrian Smith，我现在了解到r/fold在并行运行之前需要有一个可折叠的序列。

Comments

具体来说，你需要实现 CollFold 的东西。默认情况下，向量和一些其他类型（范围、迭代、哈希表）实现了它，但默认回退是使用串行 reduce。简单的想法是使用向量，稍微复杂一点的想法是定义自己的可折叠或实现 CollFold（使用 reducer api 或 reify）。

Answer 2

这是一个非常好的问题！对于这类数据的有效操作超级重要。

为此，我们开发了一些相关的库。

这个 Gist 展示了我们如何简化处理 CSV 数据的一些方法，并将你的原始合并求解方法与一个针对数据/数据类型感知并行化版本的合并求解方法进行了比较（正如预期的那样，要快得多）。

https://gist.github.com/harold/7335b78606f8e962f2b385f1ed79d15c

希望这能有所帮助，并展示了一种处理这类问题的不同方法。

PS. 从你的问题中了解了 fnil，很酷！

Comments

感谢你的 informative answer. Creating the dataset appears to take the bulk of the time (approx 82 secs from 100 M lines), whereas the processing itself is very fast - 100M lines in about 10 secs.

---

当然，欢迎您。是的，100M 是很多行！解析 200M 个数字肯定会花一点时间。您的问题涉及到并行化合并/求和，所以我的代码将这部分工作分离开。如果您想优化数据集的保存和加载，`nippy` 通常比 CSV 快得多。请查看 `tech.io/put-nippy!` 和 `tech.io/get-nippy`。

---

谢谢。尝试使用 (ds/write-csv! (ds/->dataset source-data) 或 (io/put-nippy! "./data.nippy" (ds/->dataset source-data))) 来生产 100M 行，导致我的机器堆栈溢出。我是用原始代码生产 csv（并手动添加标题）的。

---

我明白了。这很有道理，并且是 `clojure.data.csv` 流式处理能力的酷例子。这台机器有 32GB 的 RAM，所以我能够修改我的代码，写入 100M 行的 csv 和 nippy 数据集。磁盘上的 csv 大约 658MB，nippy 大约 537MB。csv 数据集加载需要约 47 秒，nippy 数据集加载需要约 2.8 秒。(