将图分布到集群节点

Question

我在 Dask.delayed 方面取得了不错的进展。作为一个团队，我们决定将更多时间用于使用 Dask 处理图表。

我有一个关于分发的问题。我在我们的集群上看到以下行为。我启动例如8 个节点上的每个节点都有 8 个工作人员，每个节点有 4 个线程，比如说/我然后 client.compute 8 个图表来创建模拟数据以供后续处理。我想让 8 个数据集为每个节点生成一个。但是，似乎发生的情况并非没有道理，这八个功能在前两个节点上运行。随后的计算在第一和第二节点上运行。因此，我看到缺乏缩放。随着时间的推移，其他节点将从诊断工作者页面中消失。这是预期的吗？

所以我想先按节点分配数据创建功能。所以当我想计算图表时，我现在这样做：

if nodes is not None:
    print("Computing graph_list on the following nodes: %s" % nodes)
    return client.compute(graph_list, sync=True, workers=nodes, **kwargs)
else:
    return client.compute(graph_list, sync=True, **kwargs)

这似乎设置正确：诊断进度条显示我的数据创建功能在内存中，但它们没有启动。如果省略节点，则计算按预期进行。此行为同时出现在集群和我的桌面上。

更多信息：查看调度程序日志，我确实看到了通信故障。

more dask-ssh_2017-09-04_09\:52\:09/dask_scheduler_sand-6-70\:8786.log
distributed.scheduler - INFO - -----------------------------------------------
distributed.scheduler - INFO -   Scheduler at:    tcp://10.143.6.70:8786
distributed.scheduler - INFO -       bokeh at:              0.0.0.0:8787
distributed.scheduler - INFO -        http at:              0.0.0.0:9786
distributed.scheduler - INFO - Local Directory:    /tmp/scheduler-ny4ev7qh
distributed.scheduler - INFO - -----------------------------------------------
distributed.scheduler - INFO - Register tcp://10.143.6.73:36810
distributed.scheduler - INFO - Starting worker compute stream, tcp://10.143.6.73:36810
distributed.scheduler - INFO - Register tcp://10.143.6.71:46656
distributed.scheduler - INFO - Starting worker compute stream, tcp://10.143.6.71:46656
distributed.scheduler - INFO - Register tcp://10.143.7.66:42162
distributed.scheduler - INFO - Starting worker compute stream, tcp://10.143.7.66:42162
distributed.scheduler - INFO - Register tcp://10.143.7.65:35114
distributed.scheduler - INFO - Starting worker compute stream, tcp://10.143.7.65:35114
distributed.scheduler - INFO - Register tcp://10.143.6.70:43208
distributed.scheduler - INFO - Starting worker compute stream, tcp://10.143.6.70:43208
distributed.scheduler - INFO - Register tcp://10.143.7.67:45228
distributed.scheduler - INFO - Starting worker compute stream, tcp://10.143.7.67:45228
distributed.scheduler - INFO - Register tcp://10.143.6.72:36100
distributed.scheduler - INFO - Starting worker compute stream, tcp://10.143.6.72:36100
distributed.scheduler - INFO - Register tcp://10.143.7.68:41915
distributed.scheduler - INFO - Starting worker compute stream, tcp://10.143.7.68:41915
distributed.scheduler - INFO - Receive client connection: 5d1dab2a-914e-11e7-8bd1-180373ff6d8b
distributed.scheduler - INFO - Worker 'tcp://10.143.6.71:46656' failed from closed comm: Stream is clos
ed
distributed.scheduler - INFO - Remove worker tcp://10.143.6.71:46656
distributed.scheduler - INFO - Removed worker tcp://10.143.6.71:46656
distributed.scheduler - INFO - Worker 'tcp://10.143.6.73:36810' failed from closed comm: Stream is clos
ed
distributed.scheduler - INFO - Remove worker tcp://10.143.6.73:36810
distributed.scheduler - INFO - Removed worker tcp://10.143.6.73:36810
distributed.scheduler - INFO - Worker 'tcp://10.143.6.72:36100' failed from closed comm: Stream is clos
ed
distributed.scheduler - INFO - Remove worker tcp://10.143.6.72:36100
distributed.scheduler - INFO - Removed worker tcp://10.143.6.72:36100
distributed.scheduler - INFO - Worker 'tcp://10.143.7.67:45228' failed from closed comm: Stream is clos
ed
distributed.scheduler - INFO - Remove worker tcp://10.143.7.67:45228
distributed.scheduler - INFO - Removed worker tcp://10.143.7.67:45228
(arlenv) [hpccorn1@login-sand8 performance]$

这会引发任何可能的原因吗？

谢谢， 蒂姆

Answer 1

Dask 如何选择将任务分配给工作人员是很复杂的，并且会考虑负载平衡、数据传输、资源限制等许多问题。如果没有具体而简单的信息，就很难推断事情最终会走向何方例子。

您可以尝试的一件事是一次提交所有计算，这可以让调度程序做出更明智的决策，而不是一次只看到一个。

所以，你可以试试这样替换代码：

futures = [client.compute(delayed_value) for delayed_value in L]
wait(futures)

这样的代码

futures = client.compute(L)
wait(futures)

但老实说，我只给了 30% 的机会来解决您的问题。如果不深入研究您的问题，就很难知道发生了什么。如果您能提供一个非常简单可重现的代码示例，那将是最好的。