Killed/MemoryError 从延迟收集创建大型 dask.dataframe 时
Killed/MemoryError when creating a large dask.dataframe from delayed collection
我正在尝试从一堆大型 CSV 文件（目前有 12 个文件，每个文件有 8-10 百万行和 50 列）创建一个 dask.dataframe。它们中的一些可能会一起放入我的系统内存中，但它们肯定不会同时全部放入，因此使用 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 的全部意义在于能够对跨越磁盘上多个文件的大于内存的数据帧进行操作，那么我在这里做错了什么？
edit: 据我所知，用 df = dd.read_csv(path + '/*.csv') 读取文件似乎工作正常。但是，这不允许我使用文件路径中的附加数据来更改每个数据帧。
编辑#2：
按照 MRocklin 的回答，我尝试使用 dask 的 read_bytes() method as well as using the single-threaded scheduler 读取我的数据，并将两者结合起来。
尽管如此，即使在具有 8GB 内存的笔记本电脑上以单线程模式读取 100MB 的块，我的进程迟早会被杀死。 运行 下面针对一堆形状相似的小文件（每个大约 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))
如果您的每个文件都很大，那么在 Dask 有机会变聪明之前，对 read_file_to_dataframe 的几个并发调用可能会淹没内存。
Dask 尝试通过 运行 函数在低内存中运行，以便它可以快速删除中间结果。然而，如果只有几个函数的结果可以填满内存，那么 Dask 可能永远没有机会删除东西。例如，如果您的每个函数都生成一个 2GB 的数据帧，并且如果您同时有八个线程 运行，那么您的函数可能会在 Dask 的调度策略生效之前生成 16GB 的数据。
一些选项
使用dask.bytes.read_bytes
read_csv 起作用的原因是它将大型 CSV 文件分成许多约 100MB 的字节块（请参阅 blocksize= 关键字参数）。您也可以这样做，尽管这很棘手，因为您需要始终在端线处中断。
dask.bytes.read_bytes函数可以帮到你。它可以将单个路径转换为 delayed objects 的列表，每个对应于该文件的一个字节 运行ge，在分隔符上干净地开始和停止。然后，您可以将这些字节放入 io.BytesIO（标准库）并对其调用 pandas.read_csv。请注意，您还必须处理 headers 等。该函数的文档字符串很广泛，应该提供更多帮助。
使用单线程
在上面的示例中，如果我们没有来自并行性的 8 倍乘数，一切都会很好。我怀疑如果你一次只 运行 一个函数，那么事情可能会在没有达到你的内存限制的情况下流水线。您可以将 dask 设置为仅使用具有以下行的单个线程
dask.set_options(get=dask.async.get_sync)
注意: For Dask versions >= 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 等数据格式。
我正在尝试从一堆大型 CSV 文件（目前有 12 个文件，每个文件有 8-10 百万行和 50 列）创建一个 dask.dataframe。它们中的一些可能会一起放入我的系统内存中，但它们肯定不会同时全部放入，因此使用 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 的全部意义在于能够对跨越磁盘上多个文件的大于内存的数据帧进行操作，那么我在这里做错了什么？
edit: 据我所知，用 df = dd.read_csv(path + '/*.csv') 读取文件似乎工作正常。但是，这不允许我使用文件路径中的附加数据来更改每个数据帧。
编辑#2： 按照 MRocklin 的回答，我尝试使用 dask 的 read_bytes() method as well as using the single-threaded scheduler 读取我的数据，并将两者结合起来。 尽管如此，即使在具有 8GB 内存的笔记本电脑上以单线程模式读取 100MB 的块，我的进程迟早会被杀死。 运行 下面针对一堆形状相似的小文件（每个大约 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))
如果您的每个文件都很大，那么在 Dask 有机会变聪明之前，对 read_file_to_dataframe 的几个并发调用可能会淹没内存。
Dask 尝试通过 运行 函数在低内存中运行，以便它可以快速删除中间结果。然而，如果只有几个函数的结果可以填满内存，那么 Dask 可能永远没有机会删除东西。例如，如果您的每个函数都生成一个 2GB 的数据帧，并且如果您同时有八个线程 运行，那么您的函数可能会在 Dask 的调度策略生效之前生成 16GB 的数据。
一些选项
使用dask.bytes.read_bytes
read_csv 起作用的原因是它将大型 CSV 文件分成许多约 100MB 的字节块（请参阅 blocksize= 关键字参数）。您也可以这样做，尽管这很棘手，因为您需要始终在端线处中断。
dask.bytes.read_bytes函数可以帮到你。它可以将单个路径转换为 delayed objects 的列表，每个对应于该文件的一个字节 运行ge，在分隔符上干净地开始和停止。然后，您可以将这些字节放入 io.BytesIO（标准库）并对其调用 pandas.read_csv。请注意，您还必须处理 headers 等。该函数的文档字符串很广泛，应该提供更多帮助。
使用单线程
在上面的示例中，如果我们没有来自并行性的 8 倍乘数，一切都会很好。我怀疑如果你一次只 运行 一个函数，那么事情可能会在没有达到你的内存限制的情况下流水线。您可以将 dask 设置为仅使用具有以下行的单个线程
dask.set_options(get=dask.async.get_sync)
注意: For Dask versions >= 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 等数据格式。