为什么 dask.distributed.Client 在提供一个已使用定义的 LocalCluster 参数时，会引发 "TypeError: cannot pickle '_thread.RLock' object"？

Question

我能够创建一个 dask.distributed.LocalCluster 的空参数构造函数来隐式地使用 dask.distributed.Client. 然而，当我试图用关键字参数显式定义集群时，它失败了，并引发了异常。TypeError: cannot pickle '_thread.RLock' object.

下面是一个简单的重现器。

import dask.bag
from dask.distributed import Client, LocalCluster


def main():
    with Client():
        dask.bag.from_sequence(range(5)).map(print).compute()  # prints 0 through 4

    cluster = LocalCluster()
    with Client(cluster=cluster):  # raises TypeError: cannot pickle '_thread.RLock' object
        dask.bag.from_sequence(range(5)).map(print).compute()


if __name__ == "__main__":
    main()

Answer 1

问题是您向该函数提供了错误的关键字参数。Client 构造函数。cluster 您提供的关键字arg会被解释为应该传递给工人的关键字参数。由于 LocalCluster 不能被腌制，这将产生你所看到的异常。

Client(cluster=cluster) 应该被替换为 Client(address=cluster) 或干脆 Client(cluster).

那么一个完整的工作实例是：

import dask.bag
from dask.distributed import Client, LocalCluster


def main():
    cluster = LocalCluster()
    with Client(address=cluster):
        dask.bag.from_sequence(range(5)).map(print).compute()


if __name__ == "__main__":
    main()