python - 从延迟收集创建大型 dask.dataframe 时被杀死/内存错误

Question

我正在尝试创建一个 dask.dataframe来自一堆大型 CSV 文件(目前有 12 个文件，每个文件有 8-10 百万行和 50 列)。它们中的一些可能会一起放入我的系统内存中，但它们肯定不会同时全部放入，因此使用 dask 而不是常规 pandas。

由于读取每个 csv 文件涉及一些额外的工作(添加包含文件路径中的数据的列)，我尝试从延迟对象列表创建 dask.dataframe，类似于 to this example .

这是我的代码:

import dask.dataframe as dd
from dask.delayed import delayed
import os
import pandas as pd

def read_file_to_dataframe(file_path):
    df = pd.read_csv(file_path)
    df['some_extra_column'] = 'some_extra_value'
    return df

if __name__ == '__main__':
    path = '/path/to/my/files'
    delayed_collection = list()
    for rootdir, subdirs, files in os.walk(path):
        for filename in files:
            if filename.endswith('.csv'):
                file_path = os.path.join(rootdir, filename)
                delayed_reader = delayed(read_file_to_dataframe)(file_path)
                delayed_collection.append(delayed_reader)

    df = dd.from_delayed(delayed_collection)
    print(df.compute())

启动此脚本(Python 3.4，dask 0.12.0)时，它会运行几分钟，而我的系统内存会不断填满。当它被完全使用时，一切都开始滞后，它会再运行几分钟，然后它会因 killed 或 MemoryError 而崩溃。

我认为 dask.dataframe 的全部意义在于能够对跨越磁盘上多个文件的大于内存的数据帧进行操作，那么我在这里做错了什么？

编辑: 据我所知，使用 df = dd.read_csv(path + '/*.csv') 读取文件似乎工作正常。但是，这不允许我使用文件路径中的附加数据来更改每个数据帧。

编辑#2:按照 MRocklin 的回答，我尝试使用 dask 的 read_bytes() method 读取我的数据。以及使用 single-threaded scheduler以及两者结合使用。尽管如此，即使在具有 8GB 内存的笔记本电脑上以单线程模式读取 100MB 的 block ，我的进程迟早会被杀死。不过，在一堆形状相似的小文件(每个大约 1MB)上运行下面所述的代码效果很好。知道我在这里做错了什么吗？

import dask
from dask.bytes import read_bytes
import dask.dataframe as dd
from dask.delayed import delayed
from io import BytesIO
import pandas as pd

def create_df_from_bytesio(bytesio):
    df = pd.read_csv(bytesio)
    return df

def create_bytesio_from_bytes(block):
    bytesio = BytesIO(block)
    return bytesio


path = '/path/to/my/files/*.csv'

sample, blocks = read_bytes(path, delimiter=b'\n', blocksize=1024*1024*100)
delayed_collection = list()
for datafile in blocks:
    for block in datafile:
        bytesio = delayed(create_bytesio_from_bytes)(block)
        df = delayed(create_df_from_bytesio)(bytesio)
        delayed_collection.append(df)

dask_df = dd.from_delayed(delayed_collection)
print(dask_df.compute(get=dask.async.get_sync))

Answer 1

如果您的每个文件都很大，那么在 Dask 有机会变聪明之前，对 read_file_to_dataframe 的几个并发调用可能会淹没内存。

Dask 尝试通过按顺序运行函数来在低内存中运行，以便快速删除中间结果。然而，如果只有几个函数的结果可以填满内存，那么 Dask 可能永远没有机会删除东西。例如，如果您的每个函数都生成了一个 2GB 的数据帧，并且您同时运行了 8 个线程，那么在 Dask 的调度策略生效之前，您的函数可能会生成 16GB 的数据。

一些选项

使用 dask.bytes.read_bytes

read_csv 起作用的原因是它将大的 CSV 文件分成许多约 100MB 的字节 block (参见 blocksize= 关键字参数)。您也可以这样做，尽管这很棘手，因为您需要始终在端线处中断。

dask.bytes.read_bytes 函数可以帮到你。它可以将单个路径转换为 ​​delayed 对象列表，每个对象对应于该文件的一个字节范围，该字节范围在分隔符上干净地开始和停止。然后，您可以将这些字节放入 io.BytesIO(标准库)并调用 pandas.read_csv。请注意，您还必须处理 header 等。该函数的文档字符串非常广泛，应该会提供更多帮助。

使用单线程

在上面的示例中，如果我们没有来自并行性的 8 倍乘数，一切都会很好。我怀疑如果你一次只运行一个函数，那么事情可能会在没有达到你的内存限制的情况下流水线。您可以使用以下行将 dask 设置为仅使用单个线程

dask.set_options(get=dask.async.get_sync)

注意:For Dask 版本 >= 0.15，您需要改用 dask.local.get_sync。

确保结果适合内存(对编辑 2 的响应)

如果你制作一个 dask.dataframe 然后立即计算它

ddf = dd.read_csv(...)
df = ddf.compute()

您正在将所有数据加载到 Pandas 数据框中，这最终会耗尽内存。相反，最好在 Dask 数据帧上操作并且只计算小的结果。

# result = df.compute()  # large result fills memory
result = df.groupby(...).column.mean().compute()  # small result

转换为不同的格式

CSV 是一种普遍实用的格式，但也有一些缺陷。您可能会考虑使用 HDF5 或 Parquet 等数据格式。