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

欢迎!请参阅关于页面了解更多关于如何工作的信息。

0投票
Sequences
编辑

我本以为以下操作将会并行化运行在所有可用的CPU(在本例中是12个)上,但它只在其中一个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})

2 个答案

+1 投票

选定
 
最佳答案

感谢Clojurians Slack频道的Sean Corfield和Adrian Smith,我现在明白r/fold需要可折叠的序列才能并行运行。

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

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

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

这个 gist 展示了我们是如何简化处理 CSV 数据的,还比较了一个简单解决方案与您用于数据集/数据类型感知的并行版本(这意料之外地要快得多)。

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

希望这有助于理解,并展示了一种处理此类问题的不同方法。

PS. 从你的问题中学到了关于 fnil 的知识,很酷!

感谢你的信息丰富的回答。创建数据集似乎花费了大部分时间(大约从 1 亿行中用了 82 秒),而 processing 本身非常快 - 1 亿行花了大约 10 秒。
by
编辑了 by
当然,欢迎你。是的,100M行确实很多!解析200M数字肯定需要一些时间。你的问题是关于合并/求和的并行化,所以我将这部分工作分离出来。如果你想优化数据集的保存和加载,`nippy`通常比CSV快得多。查看`tech.io/put-nippy!`和`tech.io/get-nippy`。
by
谢谢。尝试生成100M行数据,使用`(ds/write-csv! (ds/->dataset source-data))`或`(io/put-nippy! "./data.nippy" (ds/->dataset source-data))`使我的机器的堆栈溢出。我使用原始代码生成了csv文件(并手动添加了标题)。
by
明白了。这很有道理,这是`clojure.data.csv`流式处理能力的另一个酷例子。这台机器有32GB的RAM,所以我能够修正我的代码来写入1亿行的csv和nippy数据集。磁盘上的csv文件大约是658MB,nippy文件大约是537MB。csv数据集加载大约需要47秒,nippy数据集加载大约需要2.8秒。
...