python-3.x - Dask - 如何取消和重新提交停滞的任务？

Question

我经常遇到一个问题，即 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 和所有最新版本(截至 11/12)的 Dask 和 Distributed 以及所需的软件包(例如，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. 取消停滞的任务？

Answer 1

is it possible to set a general timeout for all tasks handled by the scheduler?

不幸的是，截至 2019-11-13，答案是否定的。

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

如果 worker 宕机，那么事情会在别处尝试。然而，从你所说的听起来一切都很健康，只是任务本身可能会永远持续下去。不幸的是，很难将其确定为失败案例。