Clojure : Group-by too slow (13 million-lines file)

Question

情况

我有一个 1300 万行的 CSV，我想对每个组执行逻辑回归(incanter)。
我的文件就是这样(值只是示例)

ID Max Probability
1  1   0.5 
1  5   0.6
1  10  0.99
2  1   0.1
2  7   0.95

所以我首先用 csv 阅读器阅读了它，一切都很好。

我有这样的事情:

( {"Id" "1", "Max" 1, "Probability" 0.5} {"Id" "1", "Max" 5, "Probability" 0.6} etc.

我想按 Id 对这些值进行分组，如果我没记错的话，大约有 120 万个 Id。 (我用 Pandas 在 Python 中做到了，而且速度非常快)

这是我读取和格式化文件的功能(它适用于较小的数据集):

  (defn read-file
  []
    (let [path (:path-file @config)
          content-csv (take-csv path \,)]
      (->> (group-by :Id content-csv)
           (map (fn [[k v]]
                [k {:x (mapv :Max v) :y (mapv :Probability v)}]))
           (into {}))))

我最终希望有类似的东西来执行逻辑回归(我对此很灵活，不需要 :x 和 :y 的向量，seqs 可以)

{"1" {:x [1 5 10] :y [0.5 0.6 0.99]} "2" {:x [1 7] :y [0.1 0.95]} etc.

问题

我在按分组操作时遇到问题。我在 CSV 的输出上单独尝试了它，当它不会因为 Java 堆空间内存而消失时，这将永远持续下去。
我认为问题出在我的 mapv 上，但这是 group-by。

我想过使用 reduce 或 reduce-kv 但我不知道如何将这些功能用于此类目的。

我不关心 ":x"和 ":y"的顺序(只要它们之间相同，我的意思是 x 和 y 具有相同的索引......不是问题，因为它们在同一个行)和最终结果的 Id，我读了那个 group-by 保持顺序。
也许这就是手术成本高昂的地方？

如果有人遇到过以下情况，我会为您提供示例数据:

(def sample '({"Id" "1" "Max" 1 "Probability" 0.5} {"Id" "1" "Max" 5 "Probability" 0.6} {"Id" "1" "Max" 10 "Probability" 0.99} {"Id" "2" "Max" 1 "Probability" 0.1} {"Id" "2" "Max" 7 "Probability" 0.95}))

其他选择

我有其他想法，但我不确定它们是否适合“Clojure”。

在 Python 中，由于函数的性质和文件已经排序，我没有使用 group-by，而是在每个组的数据帧开始和结束索引中写入，这样我只需要直接选择子数据选项卡。

我还可以加载一个 id 列表，而不是从 Clojure 计算它。
喜欢

(def ids '("1""2"等

所以也许可以从以下开始:

{"1" {:x [] :y []} "2" {:x [] :y []} etc.

从前一个序列，然后匹配每个 ID 上的大文件。

我不知道它实际上是否更有效。

我有逻辑回归的所有其他功能，我只是缺少这部分!
谢谢 !

编辑

感谢您的回答，我终于有了这个解决方案。

在我的 project.clj 文件中

 :jvm-opts ["-Xmx13g"])

代码 :

(defn data-group->map [group]
  {(:Id (first group))
   {:x (map :Max group)
    :y (map :Probability group)}})


(defn prob-cumsum [data]
  (cag/fmap
    (fn [x]
      (assoc x :y (reductions + (x :y))))
  data))


(defn process-data-splitter [data]
  (->> (partition-by :Id data)
       (map data-group->map)
       (into {})
       (prob-cumsum)))

我包装了我所有的代码并且它有效。拆分大约需要 5 分钟，但我不需要超速。内存使用量可以上升到用于文件读取的所有内存，然后是 sigmoid。

Answer 1

如果您的文件按 id 排序，则可以使用 partition-by而不是 group-by .

那么您的代码将如下所示:

(defn data-group->map [group]
  [(:Id (first group))
   {:x (mapv :Max group)
    :y (mapv :Probability group)}])

(defn read-file []
  (let [path (:path-file @config)
        content-csv (take-csv path \,)]
    (->> content-csv
         (partition-by :Id)
         (map data-group->map)
         (into {}))))

这应该加快速度。
然后你可以使用换能器让它更快

(defn read-file []
  (let [path (:path-file @config)
        content-csv (take-csv path \,)]
    (into {} (comp (partition-by :Id)
                   (map data-group->map))
          content-csv)))

让我们做一些测试:

首先生成像你这样的巨大数据:

(def huge-data
  (doall (mapcat #(repeat 
                     1000000
                     {:Id % :Max 1 :Probability 10})
           (range 10))))

我们有千万项数据集，百万条 {:Id 0 :Max 1 :Probability 10} ，万 {:Id 1 :Max 1 :Probability 10}等等。

现在要测试的功能:

(defn process-data-group-by [data]
  (->> (group-by :Id data)
       (map (fn [[k v]]
              [k {:x (mapv :Max v) :y (mapv :Probability v)}]))
       (into {})))

(defn process-data-partition-by [data]
  (->> data
       (partition-by :Id)
       (map data-group->map)
       (into {})))

(defn process-data-transducer [data]
  (into {} (comp (partition-by :Id) (map data-group->map)) data))

现在时间测试:

(do (time (dorun (process-data-group-by huge-data)))
    (time (dorun (process-data-partition-by huge-data)))
    (time (dorun (process-data-transducer huge-data))))

"Elapsed time: 3377.167645 msecs"
"Elapsed time: 3707.03448 msecs"
"Elapsed time: 1462.955152 msecs"

注意， partition-by产生惰性序列，而 group-by 应该实现整个集合。因此，如果您需要逐组的数据，而不是整个 map ，则可以删除 (into {})并更快地访问每一个:

(defn process-data-partition-by [data]
  (->> data
       (partition-by :Id)
       (map data-group->map)))

查看:

user> (time (def processed-data (process-data-partition-by huge-data)))
"Elapsed time: 0.06079 msecs"
#'user/processed-data
user> (time (let [f (first processed-data)]))
"Elapsed time: 302.200571 msecs"
nil
user> (time (let [f (second processed-data)]))
"Elapsed time: 500.597153 msecs"
nil
user> (time (let [f (last processed-data)]))
"Elapsed time: 2924.588625 msecs"
nil
user.core> (time (let [f (last processed-data)]))
"Elapsed time: 0.037646 msecs"
nil