2024 Clojure状态调查!中分享您的想法。

欢迎!请在《关于》页面了解更多关于这个工作方式的信息。

0
序列
编辑

我原本以为下面的代码会在所有可用的cpu(在此情况下为12个)上并行运行,但实际上它只有一个运行。我错在哪里了?

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

2个答案

+1

选定
 
最佳答案

感谢Clojurians聊天室的Sean Corfield和Adrian Smith,我现在明白r/fold需要一个可折叠的序列才能并行运行。

具体来说,你需要实现 CollFold 的东西。默认情况下,向量和一些其他类型(范围、迭代器、哈希表)实现了它,但默认回退是使用串行归约。所以简单来说是使用向量,稍微复杂一点的是定义你自己的可折叠类型或实现 CollFold(使用归约器 API 或重新实现)。
0

这是一个非常好的问题!对此类数据的有效操作非常重要。

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

这份代码片段显示了我们是如何简化处理 csv 数据的,以及将您所做的数据集/数据类型感知的并行化方法与简单的求和/合并方法进行了比较(不出所料,这个版本要快得多)。

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

希望这能帮到您,并展示了处理此类问题的一种不同的方法。

PS。从您的问题中了解到 fnil,很酷!

感谢您富有信息量的回答。创建数据集似乎需要大部分时间(约 82 秒,从 1 亿行),而处理本身非常快 - 大约 10 秒可以处理 1 亿行。

编辑了
当然,您随时欢迎。是的,100M 行确实不少!解析 2 亿个数字肯定需要一些时间。您的问题是关于并行合并/求和,所以我的代码将这部分工作分离出来。如果您想优化数据集的保存和加载,`nippy` 通常比 CSV 快得多。请查看 `tech.io/put-nippy!` 和 `tech.io/get-nippy`。
by
谢谢。尝试使用 `(ds/write-csv! (ds/->dataset source-data) 或 (io/put-nippy! "./data.nippy" (ds/->dataset source-data)))` 产生 1000 万行数据会使我机器的堆栈溢出。我用原始代码生成了 csv(并手动添加了标题)。
by
明白了。这很有道理,是 `clojure.data.csv` 的流式处理可能性的一个酷例子。这台机器有 32GB 内存,所以我修改了代码以写入 1000 万行 CSV 和 nippy 数据集。磁盘上的 csv 文件大小约为 658MB,nippy 的大小约为 537MB。csv 数据集加载大约需要 47 秒,nippy 数据集加载大约需要 2.8 秒。
...