使用 Cassava 在内存中加载 CSV

Question

我正在尝试将 CSV 作为带有 Cassava 的 Vector 加载到内存中。我的程序确实可以工作，但是对于 50MB 的 csv 文件使用了大量内存，我不明白为什么。

我知道使用 Data.Csv.Streaming 应该更适合大文件，但我认为 50MB 仍然可以。我尝试了 Data.Csv 和 Data.Csv.Streaming 以及来自 github 项目页面的或多或少的规范示例，我还尝试实现自己的解析器，输出 Vector of Vector（我的代码基于 attoparsec-csv https://hackage.haskell.org/package/attoparsec-csv） ，所有这些解决方案都使用大约 2000MB 的内存！我确信我正在做的事情有问题。这样做的正确方法是什么？

我的最终目标是将数据完全加载到内存中以供以后进一步处理。例如，我可以将我的数据拆分为有趣的矩阵，并与使用 Hmatrix 的矩阵一起工作。

这是我用 Cassava 尝试过的 2 个程序：

1/ 使用 Data.Csv

import qualified Data.ByteString.Lazy as BL
import qualified Data.Vector as V
import Data.Csv
import Data.Foldable


main = do
   csv <- BL.readFile "train.csv"
   let Right res = decode HasHeader csv :: Either String (V.Vector(V.Vector(BL.ByteString)))
   print $ res V.! 0

2/ 使用 Data.Csv.Streaming

{-# LANGUAGE BangPatterns #-}

import qualified Data.ByteString.Lazy as BL
import qualified Data.Vector as V
import Data.Csv.Streaming
import Data.Foldable


main = do
   csv <- BL.readFile "train.csv"
   let !a = decode HasHeader csv :: Records(V.Vector(BL.ByteString))
   let !res = V.fromList $ Data.Foldable.toList a
   print $ res V.! 0

请注意，我没有给你我基于 attoparsec-csv 制作的程序，因为它与 Vector 而不是 List 几乎完全相同。该解决方案的内存使用率仍然很差。

有趣的是，在 Data.Csv.Streaming 解决方案中，如果我简单地使用 Data.Foldable.for_ 打印我的数据，一切都非常快，内存使用量为 2MB。这让我觉得我的问题与我构建 Vector 的方式有关。可能是累积 thunk 而不是将原始数据堆叠成紧凑的数据结构。

感谢您的帮助，

安东尼

Answer 1

Data.CSV 和 Data.CSV.Streaming 之间的区别可能并不完全符合您的预期。如您所见，第一个生成 csv 内容的Data.Vector.Vector。我不确定为什么这个向量的构造会占用这么多空间——尽管当我反映生成的指针到指针到向量的向量到惰性时，它开始并不让我感到惊讶-这里的字节串包含 28203420 个指向惰性字节串的不同指针，每行 371 个，每个指向原始字节流的一小部分，通常指向“0”。在http://blog.johantibell.com/2011/06/memory-footprints-of-some-common-data.html 之后，这意味着原始字节流中典型的两个字节 序列——几乎所有这些序列看起来都像这样：“,0”，即。 [44,48] - 被许多指针和构造函数取代：仅惰性字节串内容就使每对字节占用大约 11 个字（Chunk 和 Empty 构造函数用于惰性字节串，加上用于J Tibell 放在 9 个字上的严格字节串）......加上原始字节（减去那些表示逗号和空格的字节）。在 64 位系统中，这是一个相当大的规模升级。

Data.CSV.Streaming 并没有太大的不同：基本上它构造了一个稍微修饰的列表而不是一个向量，所以原则上它可以被延迟评估，并且在理想情况下，整个事情不需要在内存中实现，正如你所注意到的。但是，在这样的单子上下文中，您将“从 IO 中提取列表”，这并不能完全保证会产生混乱和混乱。

如果您想正确地流式传输 csv 内容，您应该使用 ... 流式库之一。 （我没有建议将整个内容放入内存，除了安排 cassava 将每一行读取为一个很好的紧凑数据类型而不是指向惰性字节字符串的指针向量；这里虽然我们有 371 个“字段”）。

所以这里是你使用cassava-streams的程序，它使用了cassava的（正版）增量接口，然​​后使用io-streams做一个记录流：

  {-# LANGUAGE BangPatterns #-}

  import qualified Data.ByteString.Lazy as BL
  import qualified Data.Vector as V
  import Data.Foldable
  import System.IO.Streams (InputStream, OutputStream)
  import qualified System.IO.Streams as Streams
  import qualified System.IO.Streams.Csv as CSV
  import System.IO

  type StreamOfCSV = InputStream (V.Vector(BL.ByteString))

  main = withFile "train.csv" ReadMode $ \h -> do
     input          <- Streams.handleToInputStream h 
     raw_csv_stream <- CSV.decodeStream HasHeader input
     csv_stream     <- CSV.onlyValidRecords raw_csv_stream :: IO StreamOfCSV
     m <- Streams.read csv_stream
     print m

这会立即完成，使用不超过hello-world 的内存，打印第一条记录。您可以在教程源 https://github.com/pjones/cassava-streams/blob/master/src/System/IO/Streams/Csv/Tutorial.hs 中看到更多操作，其他流库也有类似的库。如果您要构建的数据结构（如矩阵）可以放入内存，您应该能够通过使用Streams.fold 折叠行来构建它，并且如果您尝试从中提取的信息应该没有问题每行在被折叠操作消耗之前都经过正确评估。如果您可以安排 cassava 输出带有未装箱字段的非递归数据结构，则可以为该类型编写一个 Unbox 实例，并将整个 csv 折叠成一个紧密包装的未装箱向量。在这种情况下，每行有 371 个不同的字段，所以我猜这不是一个选项。

下面是Data.CSV.Streaming程序的等价物：

  main = withFile "train.csv" ReadMode $ \h -> do
    input          <- Streams.handleToInputStream h 
    raw_csv_stream <- CSV.decodeStream HasHeader input
    csv_stream     <- CSV.onlyValidRecords raw_csv_stream :: IO StreamOfCSV
    csvs <- Streams.toList csv_stream
    print (csvs !! 0)

它有同样的麻烦，因为它在尝试找出第一个元素之前使用Streams.toList 收集巨大的列表。

-- 附录

这里值得一提的是一个管道 csv 变体，它只是将每个已解析的行手动压缩为一个未装箱的 Ints 向量（这比找到 Doubles 更容易，这是这个 csv 真正存储的内容，使用 bytestring 包中的readInt。）

import Data.ByteString (ByteString)
import qualified Data.ByteString.Char8 as B
import qualified Data.Vector as V
import qualified Data.Vector.Unboxed as U
import Data.Csv

import qualified Pipes.Prelude as P
import qualified Pipes.ByteString as Bytes
import Pipes
import qualified Pipes.Csv as Csv
import System.IO
import Control.Applicative

import qualified Control.Foldl as L

main = withFile "train.csv" ReadMode $ \h -> do
  let csvs :: Producer (V.Vector ByteString) IO ()
      csvs = Csv.decode HasHeader (Bytes.fromHandle h) >-> P.concat
      -- shamelessly reading integral part only, counting bad parses as 0
      simplify bs = case B.readInt bs of
        Nothing       -> 0
        Just (n, bs') -> n
      uvectors :: Producer (U.Vector Int) IO ()
      uvectors = csvs  >-> P.map (V.map simplify) >-> P.map (V.foldr U.cons U.empty)
  runEffect $ uvectors >-> P.print

您可以使用 foldl 库中的折叠或任何您想编写的内容来折叠行，方法是将最后一行换成类似这样的内容

  let myfolds = liftA3 (,,) (L.generalize (L.index 13))   -- the thirteenth row, if it exists
                            (L.randomN 3)   -- three random rows
                            (L.generalize L.length) -- number of rows
  (thirteen,mvs,len) <- L.impurely P.foldM myfolds uvectors

  case mvs of 
    Nothing -> return ()
    Just vs -> print (vs :: V.Vector (U.Vector Int))
  print thirteen
  print len

在这种情况下，我正在收集第 13 行、三个随机行和记录总数 - 任何数量的其他折叠都可以与这些组合。特别是，我们也可以使用L.vector 将所有行收集到一个巨大的向量中，考虑到这个 csv 文件的大小，这可能仍然是个坏主意。下面我们回到我们的起点，我们收集所有内容并打印完成的向量的第 17 行，即一种大矩阵。

  vec_vec <- L.impurely P.foldM  L.vector uvectors
  print $ (vec_vec :: V.Vector (U.Vector Int)) V.! 17

这会占用大量内存，但不会对我的小笔记本电脑造成特别大的压力。