3

我经常遇到一个问题，即 Dask 在几个任务上随机停止，通常与从我的网络上的不同节点读取数据有关（下面有更多详细信息）。这可能在运行脚本几个小时后没有问题发生。它将以如下所示的形式无限期挂起（否则此循环需要几秒钟才能完成）：

在此处输入图像描述

在这种情况下，我看到只有少数停滞的进程，并且都在一个特定的节点（192.168.0.228）上： 在此处输入图像描述

该节点上的每个工作人员都在几个 read_parquet 任务上停滞不前：

在此处输入图像描述

这是使用以下代码调用的，并且使用的是 fastparquet：

ddf = dd.read_parquet(file_path, columns=['col1', 'col2'], index=False, gather_statistics=False)

我的集群正在运行 Ubuntu 19.04 和 Dask 和 Distributed 的所有最新版本（截至 11/12）以及所需的软件包（例如，tornado、fsspec、fastparquet 等）

.228 节点尝试访问的数据位于我集群中的另一个节点上。.228 节点通过 CIFS 文件共享访问数据。我在运行脚本的同一节点上运行 Dask 调度程序（不同于 .228 节点和数据存储节点）。该脚本使用 Paramiko 通过 SSH 将工作人员连接到调度程序：

ssh_client = paramiko.SSHClient()
stdin, stdout, stderr = ssh_client.exec_command('sudo dask-worker ' +
                                                            ' --name ' + comp_name_decode +
                                                            ' --nprocs ' + str(nproc_int) +
                                                            ' --nthreads 10 ' +
                                                            self.dask_scheduler_ip, get_pty=True)

.228 节点与调度程序和数据存储节点的连接看起来都很健康。.228 节点可能会在尝试处理 read_parquet 任务时遇到某种短暂的连接问题，但如果发生这种情况，那么 .228 节点与调度程序和 CIFS 共享的连接在该短暂时刻之后不会受到影响。在任何情况下，日志都不会显示任何问题。这是来自 .228 节点的整个日志：

distributed.worker - INFO - Start worker at: tcp://192.168.0.228:42445

distributed.worker - INFO - Listening to: tcp://192.168.0.228:42445

distributed.worker - INFO - dashboard at: 192.168.0.228:37751

distributed.worker - INFO - Waiting to connect to: tcp://192.168.0.167:8786

distributed.worker - INFO - -------------------------------------------------

distributed.worker - INFO - Threads: 2

distributed.worker - INFO - Memory: 14.53 GB

distributed.worker - INFO - Local Directory: /home/dan/worker-50_838ig

distributed.worker - INFO - -------------------------------------------------

distributed.worker - INFO - Registered to: tcp://192.168.0.167:8786

distributed.worker - INFO - -------------------------------------------------

撇开这是否是 Dask 或我的代码/网络中的错误不谈，是否可以为调度程序处理的所有任务设置一般超时？或者，是否可以：

  1. 识别停滞的任务，
  2. 复制一个停滞的任务并将其移动到另一个工人，并且
  3. 取消停滞的任务？
4

1 回答 1

1

是否可以为调度程序处理的所有任务设置一般超时？

不幸的是，截至 2019 年 11 月 13 日，答案是否定的。

如果任务正确失败，那么您可以重试该任务，client.retry(...)但没有自动方法让任务在一定时间后自行失败。这是您必须自己将其写入 Python 函数的内容。不幸的是，很难在另一个线程中中断 Python 函数，这也是未实现的部分原因。

如果工人倒下了，那么事情将在其他地方进行尝试。但是，从您所说的听起来一切都很健康，只是任务本身可能会永远持续下去。不幸的是，很难将其确定为失败案例。

于 2019-11-13T15:45:29.910 回答