Dask不会清理docker容器中的上下文

Question

[我们有一个Dask管道，我们基本上在其中使用LocalCluster作为进程池。也就是说，我们以LocalCluster(processes=True, threads_per_worker=1)启动集群。像这样：

dask_cluster = LocalCluster(processes=True, threads_per_worker=1)
    with Client(dask_cluster) as dask_client:
        exit_code = run_processing(input_file, dask_client, db_state).value

我们的工作流程和任务并行化在本地运行时效果很好。但是，当我们将代码复制到Docker容器（基于centos）时，处理完成，并且有时在容器退出时出现以下错误：

Traceback (most recent call last):^M
  File "/opt/rh/rh-python36/root/usr/lib64/python3.6/multiprocessing/queues.py", line 240, in _feed^M
    send_bytes(obj)^M
  File "/opt/rh/rh-python36/root/usr/lib64/python3.6/multiprocessing/connection.py", line 200, in send_bytes^M
    self._send_bytes(m[offset:offset + size])^M
  File "/opt/rh/rh-python36/root/usr/lib64/python3.6/multiprocessing/connection.py", line 404, in _send_bytes^M
    self._send(header + buf)^M
  File "/opt/rh/rh-python36/root/usr/lib64/python3.6/multiprocessing/connection.py", line 368, in _send^M
    n = write(self._handle, buf)^M
BrokenPipeError: [Errno 32] Broken pipe^M

此外，我们收到此错误的多个实例，这使我认为该错误来自废弃的工作进程。我们当前的工作原理是，这与"Docker zombie reaping problem"有某种联系，但是我们不知道如何从不完全不同的docker镜像开始进行修复，因此我们不希望这样做。

是否有一种仅使用Dask群集/客户端清除方法来解决此问题的方法？

我们有一个Dask管道，我们基本上在其中使用LocalCluster作为进程池。即我们使用LocalCluster（processes = True，threads_per_worker = 1）启动集群。像这样：dask_cluster = ...

Answer 1

您应该将集群创建为上下文管理器。实际上是启动进程的东西，而不是客户端。