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

Question

我有一个比我的记忆大得多的文本文件。我想按字典顺序对该文件的行进行排序。我知道如何手动操作：

1. 分成适合内存的块
2. 对块进行排序
3. 合并块

我想用 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 来阻止写入列名，在这个 大小写（不是很有意义）0。

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

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

编辑

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

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

请注意：

• 首先compute()，生成整个pandasonic DataFrame，
• 之后，在 Pandas 级别，您尝试对其进行排序。

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