Dask 数据帧大于内存

Question

我是 Dask 的新手，我发现它非常有用，但我有一个问题我还没有解决。

我有一个大于内存的数据集，我想从列中删除重复值。

问题在于，删除后数据集仍将大于内存。因此，需要通过文件计算结果，直接保存到磁盘中。

当然，我可以构建一个代码来手动执行此删除操作，但我想知道 Dask 是否已经实现了此功能。

这是我的代码：

from dask.distributed import Client
import dask.dataframe as dd

client = Client(memory_limit='8GB') # I've tried without this limit
data = dd.read_csv("path_to_file", dtype={
    'id': 'Int64'
}, sample=1000)
data.drop_duplicates(subset=['text'])
results = data.compute() #   <- Here is the problem
results.to_csv("pathout", index=False)

当我调用计算时，结果是一个 DataFrame pandas，在这种情况下，它比内存大。我收到了很多：

distributed.nanny - WARNING - Worker exceeded 95% memory budget. Restarting
distributed.nanny - WARNING - Restarting worker

然后执行失败并显示“KilledWorker”

编辑：

自包含示例：

import numpy as np
import pandas as pd
from dask.distributed import Client
import dask.dataframe as dd

# Creates about 2Gb of data
data = np.random.randint(0, 10000, size=(2000000,200))
pd.DataFrame(data).to_csv('test_in.csv', index=False)

# If you want to run on terminal, uncomment the next line and identy the rest of the code
# if __name__ == '__main__':

# To test, limit dask to 1Gb
client = Client(n_workers=1, memory_limit='1GB')
df = dd.read_csv('test_in.csv', blocksize='16MB')
results = df.drop_duplicates()
results.to_csv('test_out.csv', index=False)
client.close()

Answer 1

from dask.distributed import Client
import dask.dataframe as dd


client = Client(memory_limit='8GB')
data = dd.read_csv("path_to_file", dtype={'id': 'Int64'}, sample=1000)
results = data.drop_duplicates(subset=['A'])  # Don't call compute here
results.to_csv("pathout", index=False)  # Write operations automatically call compute

.compute() 将返回一个 Pandas 数据框，然后 Dask 就消失了。您可以使用 Dask 中的 .to_csv() 函数，它会为每个分区保存一个文件。

只需删除 .compute() ，如果每个分区都适合内存，它将起作用。

哦，你需要分配 .drop_duplicates() 的结果。

Answer 2

我认为你的工人被杀了，因为 drop_duplicates 将 df.npartitions 重置为 1。 尝试在前后打印 df.npartitions 以确保。

您可以尝试的一件事是 results = df.drop_duplicates(split_out=df.npartitions) 这仍然需要很长时间才能计算困难..