问题描述
我正在创建一个读取整个文件夹的函数，创建一个 Dask 数据帧，然后处理该数据帧的分区并对结果求和，如下所示：
import dask.dataframe as dd
from dask import delayed,compute
def partitions_func(folder):
df = dd.read_csv(f'{folder}/*.csv')
partial_results = []
for partition in df.partitions:
partial = another_function(partition)
partial_results.append(partial)
total = delayed(sum)(partial_results)
return total
在 partitions_func (another_function) 中调用的函数也被延迟了。
@delayed
def another_function(partition):
# Partition processing
return result
我检查过，处理过程中创建的变量都很小，所以它们应该不会引起任何问题。分区可以很大，但不能大于可用 RAM。
当我执行 partitions_func(folder) 时，进程被杀死。起初，我认为问题与有两个 delayed 有关，一个在 another_function 上，一个在 delayed(sum) 上。
从 delayed 中移除 another_function 装饰器会导致问题，因为参数是 Dask 数据帧，您无法执行 tolist() 之类的操作。我尝试从 delayed 中删除 sum，因为我认为这可能是并行化和可用资源的问题，但该进程也会被终止。
但是，我知道有 5 个分区。如果我从 total = delayed(sum)(partial_results) 中删除语句 partitions_func 并改为“手动”计算总和，则一切都按预期进行：
total = partial_results[0].compute() + partial_results[1].compute() + partial_results[2].compute() \
+ partial_results[3].compute() + partial_results[4].compute()
谢谢！
解决方法
暂无找到可以解决该程序问题的有效方法，小编努力寻找整理中！
如果你已经找到好的解决方法，欢迎将解决方案带上本链接一起发送给小编。
小编邮箱:dio#foxmail.com (将#修改为@）