使用 DASK 在集群上启动功能答案

【问题标题】：Launch function on cluster with DASK使用 DASK 在集群上启动功能
【发布时间】：2018-09-03 20:41:41
【问题描述】：

我是 DASK 的新手，想测试一下在集群上运行 DASK。集群有一个头服务器和几个其他节点。登录头服务器后，我可以通过简单的 ssh 无需密码即可进入其他节点。我想运行一个简单的函数来迭代一个大数组。该函数定义如下。就是将dt64转为numpy datetime对象。

import xarray as xr import numpy as np from dask import compute, delayed import dask.multiprocessing from datetime import datetime, timedelta def converdt64(dt64): ts = (dt64 - np.datetime64('1970-01-01T00:00:00Z')) / np.timedelta64(1, 's') return datetime.utcfromtimestamp(ts)

然后在终端上，我通过应用这个函数来迭代一个大小为 N 的 1D 数组。

values = [delayed(convertdt64)(x) for x in arraydata] results1 = compute(*values,scheduler='processes’)

这在头服务器上使用了一些内核，它可以工作，但速度很慢。然后我尝试使用客户端在集群的几个节点上启动该功能，如下所示：

from dask.distributed import Client client = Client("10.140.251.254:8786 ») results = compute(*values, scheduler='distributed’)

它根本不起作用。有一些警告和一条错误消息，如下所示。

distributed.comm.tcp - WARNING - Could not set timeout on TCP stream: [Errno 92] Protocol not available
distributed.comm.tcp - WARNING - Closing dangling stream in <TCP local=tcp://10.140.251.254:57257 remote=tcp://10.140.251.254:8786>

CancelledError: convertdt64-0205ad5e-214b-4683-b5c4-b6a2a6d8e52f

我也尝试了 dask.bag 并收到相同的错误消息。集群上的并行计算不起作用的原因可能是什么？是由于某些服务器/网络配置，还是我对 DASK 客户端的错误使用？提前感谢您的帮助！

最好的祝福

香农X

【问题讨论】：

标签： server cluster-computing dask-distributed dask-delayed

【解决方案1】：

...然后我尝试使用客户端在集群的多个节点上启动该功能，如下所示：

我在尝试在调度程序上运行任务时遇到了类似的问题。节点连接得很好。但是，尝试提交任务会导致取消。

记录的示例要么来自本地，要么来自与调度程序相同的节点。当我将客户端移动到调度程序节点时，问题就消失了。

【讨论】：