在dask内部调用dask生成进程

Question

我们有一个由许多任务组成的大型项目。我们使用一个dask图来安排每个任务。图的一个小样本如下所示。请注意，dask被设置为多处理模式。

dask_graph：

  universe: !!python/tuple [gcsstrategies.svc.business_service.UniverseService.load_universe_object, CONTEXT]
  raw_market_data: !!python/tuple [gcsstrategies.svc.data_loading_service.RDWLoader.load_market_data, CONTEXT, universe]
  raw_fundamental_data: !!python/tuple [gcsstrategies.svc.data_loading_service.RDWLoader.load_fundamental_data, CONTEXT, universe]

dask_keys: [raw_fundamental_data]

现在的任务之一，raw_fundamental_data，懒洋洋地安排使用@delay的dask任务，并使用dask.compute()运行它们。这种设计选择的原因是dask在raw_fundamental_data中计划和延迟运行的任务列表是根据运行时参数在运行时动态选择的。

我们看到的错误是：

守护进程不允许有子进程。

我们理解这是因为一个产生的过程正在试图孕育孩子。这个问题有什么解决办法吗？dask是否有任何方法允许通过daskgraph调度的任务使用@delay或其他方法来调度和懒散地运行自己的任务。

请注意，在我们的系统中有许多任务将使用多处理来运行自己的任务。因此，顺序执行不是一种选择。

Answer 1

多进程调度程序不能进行这种操作。但是，分布式调度器是(也可以很容易地在一台机器上使用分布式调度程序)。

相关的文档页面在这里：

• http://distributed.readthedocs.io/en/latest/task-launch.html
• http://dask.readthedocs.io/en/latest/scheduler-choice.html

下面是一个小例子

In [1]: from dask.distributed import Client, local_client

In [2]: def f(n):
   ...:     with local_client() as lc:
   ...:         futures = [lc.submit(lambda x: x + 1, i) for i in range(n)]
   ...:         total = lc.submit(sum, futures)
   ...:         return total.result()
   ...:     

In [3]: c = Client()  # start processes on local machine

In [4]: future = c.submit(f, 10)

In [5]: future.result()
Out[5]: 55

这将使用concurrent.futures接口来dask，而不是dask.delayed，但是您也可以使用dask.delayed。请参阅http://distributed.readthedocs.io/en/latest/manage-computation.html