dask - 如何使用 Dask 对大文本文件进行排序？

Question

我有一个比我的内存大得多的文本文件。我想按字典顺序对该文件的行进行排序。我知道如何手动完成:

分成适合内存的块

对块进行排序

合并块

我想用 dask 来做。我认为处理大量数据将是 dask 的一个用例。如何使用 Dask 对整个数据进行排序？

我的尝试

您可以执行 generate_numbers.py -n 550_000_000这将需要大约 30 分钟并生成一个 20 GB 的文件。

import dask.dataframe as dd

filename = "numbers-large.txt"

print("Create ddf")
ddf = dd.read_csv(filename, sep = ',', header = None).set_index(0)

print("Compute ddf and sort")
df = ddf.compute().sort_values(0)

print("Write")
with open("numbers-large-sorted-dask.txt", "w") as fp:
    for number in df.index.to_list():
        fp.write(f"{number}\n")

当我执行这个时，我得到

Create ddf
Compute ddf and sort
[2]    2437 killed     python dask-sort.py

我猜这个进程被杀死是因为它消耗了太多内存？

Answer 1

试试下面的代码:

import dask
import dask.dataframe as dd

inpFn = "numbers-large.txt"
outFn = "numbers-large-sorted-dask.txt"
blkSize = 500   # For test on a small file - increase it

print("Create ddf")
ddf = dd.read_csv(inpFn, header = None, blocksize=blkSize)

print("Sort")
ddf_sorted = ddf.set_index(0)

print("Write")
fut = ddf_sorted.to_csv(outFn, compute=False, single_file=True, header=None)
dask.compute(fut)
print("Stop")

注意我设置的这么低 blkSize参数仅用于测试目的。
在目标版本中，要么增加其值，要么下降，同时 blocksize=blkSize , 接受默认值。

如 set_index提供排序，无需调用 sort_values()其他细节是dask不支持这种方法。

就写作而言，我注意到你想生成一个
单例输出文件，而不是 序列 文件(每个文件一个
分区)，所以我通过了 single_file=True .

我还加了 header=None阻止写入列名，在此
case(意义不大) 0 .

最后要提到的细节是 compute=False , 所以那个 dask
生成一系列 future 的对象，而不执行它们
(计算它) - 现在。

到目前为止的所有操作都只构建了计算树，
没有它的执行。
直到现在，compute(...) 运行整个计算树。

编辑

您的代码可能由于以下原因而失败:

df = ddf.compute().sort_values(0)

请注意，您:

首先计算()，生成整个pandasonic DataFrame，

之后，在 Pandas 级别，您尝试对其进行排序。

问题可能是你电脑的内存不是
大到足以容纳compute() 的整个结果。
所以很可能你的代码此时失败了，没有任何
有机会对这个 DataFrame 进行排序。