csv - 使用 Cassava 在内存中加载 CSV

Question

我正在尝试使用木薯将 CSV 加载到内存中作为 Vector of 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 的矩阵一起工作。

以下是我用木薯试过的 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可能不是您所期望的。第一个产生 Data.Vector.Vector如您所见，csv 内容。我不知道为什么这个向量的构建应该占用这么多空间——尽管当我想到由此产生的指针向量到指针向量到懒惰时，它开始并没有让我感到惊讶-此处的字节串包含 28203420 个指向惰性字节串的不同指针，每行 371 个，每个指向原始字节流的一小部分，通常指向“0”。关注 http://blog.johantibell.com/2011/06/memory-footprints-of-some-common-data.html这意味着原始字节流中典型的两字节序列 - 几乎所有这些都看起来像这样:“，0”即。 [44,48] - 被许多指针和构造函数替换:单独的惰性字节串内容使每对字节占用大约 11 个字(用于惰性字节串的 Chunk 和 Empty 构造函数，加上用于严格字节串的 Material ) J Tibell 放置了 9 个单词)...加上原始字节(减去那些代表逗号和空格的字节)。在 64 位系统中，这是一个非常巨大的规模升级。
Data.CSV.Streaming并没有那么不同:基本上它构造了一个稍微修饰的列表而不是一个向量，所以原则上它可以被懒惰地评估，并且在理想情况下，整个事情不需要在内存中实现，正如你所注意到的。但是，在这样的 monadic 上下文中，您将“从 IO 中提取列表”，这并不能完全保证会产生困惑和困惑。

如果你想正确地流式传输 csv 内容，你应该使用......流媒体库之一。 (我没有建议将整个事情放入内存中，除了明显的一个安排，即木薯将每一行读入一个漂亮的紧凑数据类型，而不是一个指向惰性字节串的指针向量；尽管这里我们有 371 个“字段”)。

所以这是你的程序使用 cassava-streams ，它使用 cassava 的(正版)增量接口(interface)，然​​后使用 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 折叠行来构建它。如果您尝试从每一行中提取的信息在折叠操作使用之前正确评估，那么应该没有问题。如果您可以安排木薯输出带有未装箱字段的非递归数据结构，那么就可以为该类型编写一个 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在试图找出第一个元素之前收集巨大的列表。

-- 附录

这里的值(value)是一个管道 csv 变体，它只是将每个解析的行压缩成一个未装箱的向量 Int s 手动(这比查找 Doubles 更容易，这是这个 csv 真正存储的内容，使用 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

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

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

这需要大量内存，但不会给我的小笔记本电脑带来特别大的压力。