DASK计算与子期货

Question

我希望提交一个DASK的任务，将做到以下几点：

1. 使用dask.bag（def fakejob）
2. 计算图表建立一个懒DASK图表从1和保存它要实木复合地板（留下这部分，只是一个动机）

我需要为多个输入做到这一点，所以我一直在尝试使用dask.distributed的期货功能。

from dask.distributed import Client 

client = Client(processes=True) 

def fakejob(path): 
    return (
     dask.bag 
     .read_text(path) 
     .to_dataframe() 
    ) 

futures = client.map(fakejob, [input_path1, input_path2]) 

问题是我不断收到：AssertionError: daemonic processes are not allowed to have children

我试过以下this link并结束了第二个版本（均不同于第一1号线），但期货留“待定”永远。

from dask.distributed import Client 

client = Client(processes=True) 

def fakejob(path): 
    with dask.set_options(get=client.get): 
     return (
      dask.bag 
      .read_text(path) 
      .to_dataframe() 
     ) 

futures = client.map(fakejob, [input_path1, input_path2]) 

任何关于如何做到这一点的线索？

干杯。

Answer 1

奇怪的和稍微巨大的错误信息来自于尝试构建工作进程内的dask图（这是一个包），如果用client.map调用，事情最终会结束。如果您可以将整个工作流程放在该功能中，包括写入实木复合地板，并且不尝试将包裹传递给呼叫者，则您的第二次尝试将与本地客户端一起工作。

解决方案更简单。

bags = [dask.bag.read_text(path) 
     .to_dataframe() for path in [input_path1, input_path2]) 
futures = client.compute(bags) # run in background on the cluster 
client.gather(futures) # wait and get results 

这里，bags是DASK袋，即工作任务定义的列表，但还没有运行。您可以用dask.compute(*bags)取代最后两行，以获得结果而不用担心期货。