使用 Dask 将大于内存的数据帧缓存到本地磁盘

Question

我在 S3 中有一堆文件，它们包含一个大于内存的数据帧。

目前，我使用 Dask 将文件读入数据帧，使用较小的数据集执行内部连接（每次调用此函数时都会更改，而 huge_df 基本上是完整的数据集并且不会更改） ，调用compute得到一个更小的pandas数据框，然后做一些处理。例如：

huge_df = ddf.read_csv("s3://folder/**/*.part") 
merged_df = huge_df.join(small_df, how='inner', ...)
merged_df = merged_df.compute()
...other processing...

大部分时间都花在从 S3 下载文件上。我的问题是：有没有办法使用 Dask 将 S3 中的文件缓存在磁盘上，以便在随后调用此代码时，我可以从磁盘读取数据帧文件，而不是从 S3 读取？我想我不能只调用huge_df.to_csv(./local-dir/)，因为这会将huge_df 带入内存，这是行不通的。

我确信有一种方法可以结合使用其他工具和标准 Python IO 实用程序，但我想看看是否有一种方法可以使用 Dask 从 S3 下载文件内容并将它们存储在本地磁盘，而不会将所有内容都放入内存。

Answer 1

执行huge_df.to_csv 会起作用，因为它会将每个分区写入本地单独的文件，因此整个事情不会一次在内存中。

但是，为了回答具体问题，dask 使用fsspec 来管理文件操作，并且它允许local caching，例如，您可以这样做

huge_df = ddf.read_csv("simplecache::s3://folder/**/*.part")

默认情况下，这会将文件存储在一个临时文件夹中，当您退出 python 会话时，该文件夹会被清理，但您可以使用可选参数 storage_options={"simplecache": {..}} 提供选项来指定缓存位置，或使用“filecache”如果您想让本地副本在一段时间后过期或检查目标以获取更新版本，而不是“simplecache”。

请注意，显然，只有当所有工作人员都可以访问相同的缓存位置时，这些才适用于分布式集群，因为分区的加载可能发生在您的任何工作人员身上。