dask - 如何设置登录 dask 分布式 worker ？

Question

将 dask 升级到版本 1.15.0 后，我的日志记录停止工作。

我使用 logging.config.dictConfig 来初始化 python 日志记录工具，以前这些设置传播到所有工作人员。但是升级之后就不行了。

如果我在每个工作人员的每次日志调用之前都执行 dictConfig ，则它可以工作，但这不是正确的解决方案。

所以问题是它如何在我的计算图开始执行之前初始化每个工作人员的日志记录并且每个工作人员只执行一次？

更新:

这个 hack 处理了一个虚拟示例，但对我的系统没有影响:

def init_logging():
   # logging initializing happens here
   ...

client = distributed.Client()
client.map(lambda _: init_logging, client.ncores())

更新 2:

在挖掘文档后，这解决了问题:

client.run(init_logging)

所以现在的问题是:这是解决这个问题的正确方法吗？

Answer 1

从 1.15.0 版本开始，我们现在从一个干净的进程中 fork worker ，因此您可以在调用 Client() 之前对您的进程进行更改。不会影响 fork 的 worker 。更多信息搜索forkserver这里:https://docs.python.org/3/library/multiprocessing.html#contexts-and-start-methods

您使用的解决方案 Client.run在我看来很好。 Client.run 目前(从 1.15.0 版开始)是在所有当前事件的工作人员上调用函数的最佳方式。

分布式系统

值得注意的是，这里您正在设置从单个计算机上的同一进程 fork 的客户端。你上面使用的技巧在分布式环境中不起作用。如果人们遇到这个问题，询问如何在集群上下文中使用 Dask 处理日志记录，我将添加此注释。

通常 Dask 不会移动日志。相反，您用来启动 Dask 的任何机制都会处理这个问题。像 SGE/SLURM/Torque/PBS 这样的作业调度程序都可以做到这一点。 YARN/Mesos/Marathon/Kubernetes 等云系统都可以做到这一点。 dask-ssh工具执行此操作。