使用Dask将中间产物和结果写入文件

Question

我有一个数据脚本应用程序，它使用dask遍历数据库并生成一些中间产物，然后将这些中间产物组合起来生成结果。现在，我想有效地写出中间产物和结果，但正如你们在下面看到的，我发现了一种效率非常低的方法，即计算中间产物比计算结果多

import dask.bag as db
from other_functions import *

input = db.read_text(file1)
processing_parameter = parse_mapping_parameters(file2)

intermediates = []
for p in mapping_parameter:
    intermediate = input.map(lambda x: process(x, p))
    intermediates.append(intermediate)

products = intermediates.pop(0)

for intermediate in intermediates:
    products = product.products(i)

result = products.map(calc_result)

for i, intermediate in enumerate(intermediates):
    intermediate.to_textfiles(f'./data/intermediate_{i}.*.txt')

result.to_textfiles(f'./data/result.*.txt')

我看到的另一种方法是将中间产物写入文件，然后使用单独的脚本再次将它们读入内存并生成结果，但在IO方面，这也感觉效率低下。在达斯克有更好的方法吗

Answer 1

最终以这种方式解决了这个问题：

import dask.bag as db
from other_functions import *

input = db.read_text(file1)
processing_parameter = parse_mapping_parameters(file2)

to_compute = []
intermediates = []
for i, p in enumerate(mapping_parameter):
    intermediate = input.map(lambda x: process(x, p))
    to_compute.append(
        intermediate.to_textfiles(f'./data/intermediate_{i}.*.txt', compute=False)
    intermediates.append(intermediate)

products = intermediates.pop(0)
for intermediate in intermediates:
    products = product.products(i)

result = products.map(calc_result)

to_compute.append(result.to_textfiles(f'./data/result.*.txt', compute=False))
dask.compute(*to_compute)

没有意识到to_text方法有一个参数，可以让您获得一个懒惰的编写器。有一次我发现这是显而易见的。不过，我不确定这有多高效