在 Clojure 中处理大文件时如何利用 memory/performance

Question

如何在处理大数据集时序数据时利用memory/performance？

大小：~3.2G

行数：~5400 万

数据集前几行

{:ts 20200601040025269 :bid 107.526000 :ask 107.529000}
{:ts 20200601040025370 :bid 107.525000 :ask 107.529000}
{:ts 20200601040026421 :bid 107.525000 :ask 107.528000}
{:ts 20200601040026724 :bid 107.524000 :ask 107.528000}
{:ts 20200601040027424 :bid 107.524000 :ask 107.528000}
{:ts 20200601040033535 :bid 107.524000 :ask 107.527000}
{:ts 20200601040034230 :bid 107.523000 :ask 107.526000}

辅助函数

(defn lines [n filename]
  (with-open [rdr (io/reader filename)]
    (doall (take n (line-seq rdr)))))

(def dataset (into [] (lines 2000 "./data/rawdata.map")))

为了获得最佳性能，我应该尽可能多地将数据检索到内存中。然而，我的笔记本只有 16GB，当我将更多数据检索到内存中时，CPU/memory 几乎使用了 95%。

我可以在 Clojure 中对大型数据集进行更好的内存管理吗？
我可以保留一个内存缓冲区来存储数据集吗？
因为这是小内存环境下的时序数据。处理完第一批数据后，可以通过line-seq检索下一批数据。
请问这个功能是用什么数据结构实现的？

请随时发表评论。

谢谢

Answer 1

使用 deftype 创建一个带有 long ts 和 double for bid ask 的类型。如果将行字符串解析为这种类型的实例，您会发现 5400 万行的数据集应该很容易装入内存。 24 个字节的数据，加上 object header 的 8 个字节，再加上数组中约 8 个字节的引用，构成 40 个字节/记录。大约 2G 堆。

更多奇特的解决方案（用于列存储的原始数组，或用于访问打包字节缓冲区的享元）是可能的，但对于您陈述的问题参数来说不需要。

要遵循的示例代码，我手头只有 phone。

Answer 2

因为数据集仅包含 54000000 行，所以如果将数据打包到内存中，可以将此数据集放入内存。假设这是你想要做的，例如为了方便随机访问，这里提供一种方法。

您无法将其放入内存的原因可能是用于表示从文件中读取的每条记录的所有对象的开销。但是，如果将值展平到字节缓冲区中，则存储这些值所需的 space 量就不会那么大。您可以将时间戳简单地表示为每个数字一个字节，并使用某种定点表示法表示金额。这是一个快速而肮脏的解决方案。

(def fixed-pt-factor 1000)
(def record-size (+ 17 4 4))
(def max-count 54000000)

(defn put-amount [dst amount]
  (let [x (* fixed-pt-factor amount)]
    (.putInt dst (int x))))


(defn push-record [dst m]
  ;; Timestamp (convert to string and push char by char)
  (doseq [c (str (:ts m))]
    (.put dst (byte c)))
  (put-amount dst (:bid m))
  (put-amount dst (:ask m))
  dst)

(defn get-amount [src pos]
  (/ (BigDecimal. (.getInt src pos))
     fixed-pt-factor))

(defn record-count [dataset]
  (quot (.position dataset) record-size))

(defn nth-record [dataset n]
  (let [offset (* n record-size)]
    {:ts (edn/read-string (apply str (map #(->> % (+ offset) (.get dataset) char) (range 17))))
     :bid (get-amount dataset (+ offset 17))
     :ask (get-amount dataset (+ offset 17 4))}))

(defn load-dataset [filename]
  (let [result (ByteBuffer/allocate (* record-size max-count))]
    (with-open [rdr (io/reader filename)]
      (transduce (map edn/read-string) (completing push-record) result (line-seq rdr)))
    result))

然后可以使用load-dataset加载数据集，record-count获取记录数，nth-record获取第n条记录：

(def dataset (load-dataset filename))

(record-count dataset)
;; => 7

(nth-record dataset 2)
;; => {:ts 20200601040026421, :bid 107.525M, :ask 107.528M}

具体如何选择表示字节缓冲区中的值由您决定，我没有特别优化它。此示例中加载的数据集只需要大约 54000000*25 字节 = 1.35 GB，这将适合内存（尽管您可能需要调整 JVM 的一些标志...）。

如果您需要加载比这更大的文件，您可以考虑将数据放入 memory-mapped file 而不是内存中的字节缓冲区。

在 Clojure 中处理大文件时如何利用 memory/performance

How to utilize memory/performance when processing a big file in Clojure

time-series

clojure

lazy-sequences

transducer