使用 Dask 处理大型压缩 csv 文件

Question

设置是我有八个大型 csv 文件（每个 32GB），每个文件都用 Zip 压缩到 8GB 文件。我无法使用未压缩的数据，因为我想节省磁盘空间并且没有 32*8GB 剩余空间。我无法加载一个文件，例如pandas 因为它不适合内存。

我认为 Dask 是该任务的合理选择，但如果您认为它适合该目的，请随时提出不同的工具。

是否可以通过并行读取压缩文件的多个块，处理每个块并将结果保存到磁盘来使用 Dask 处理一个 8GB 压缩文件？

第一个问题是Dask does not support .zip。 This issue 建议使用dask.delayed，但我也可以将格式更改为.xz 或其他。

其次，可能与压缩格式的选择有关的是，是否可以仅并行访问压缩文件的一部分。

或者最好将每个未压缩的 csv 文件拆分为适合内存的较小部分，然后使用以下内容处理重新压缩的较小部分：

import dask.dataframe as dd

df = dd.from_csv('files_*.csv.xz', compression='xz')

目前，我更喜欢类似于第一个解决方案的东西，它似乎更精简，但我可能完全误会了，因为这个域对我来说是新的。

感谢您的帮助！

Answer 1

最简单的解决方案当然是将您的大文件流式传输到每个压缩文件中（请记住以换行符结束每个文件！），然后按照您的建议使用 Dask 加载这些文件。每个较小的文件将成为内存中的一个数据帧分区，因此只要文件足够小，您在使用 Dask 处理数据时不会耗尽内存。

这里的根本原因是，格式列表 bz2、gz 或 zip 不允许随机访问，读取数据的唯一方法是从数据的开头。 xz 是唯一允许在文件中按块压缩的格式，因此原则上可以按块加载，这与真正的随机访问不太一样。那会做你所追求的。然而，这种模式实际上与拥有单独的文件非常相似，因此不值得额外努力以阻塞模式（不是默认模式）编写文件并使用函数 dask.bytes.compression.get_xz_blocks, xz_decompress，这些函数目前未用于代码库中的任何内容。