具有分布式调度程序的dask.delayed KeyError

Question

我有一个用c编写并用ctypes包装的函数interpolate_to_particles。我想使用dask.delayed对此函数进行一系列调用。

代码在没有dask的情况下成功运行

# Interpolate w/o dask
result = interpolate_to_particles(arg1, arg2, arg3)

并使用single-threaded模式的分布式调度

# Interpolate w/ dask
from dask.distributed import Client
client = Client()
result = dask.delayed(interpolate_to_particles)(arg1, arg2, arg3)
result_c = result.compute(scheduler='single-threaded')

但是如果我打电话给

result_c = result.compute()

我得到以下KeyError：

> Traceback (most recent call last):   File
> "/path/to/lib/python3.6/site-packages/distributed/worker.py",
> line 3287, in dumps_function
>     result = cache_dumps[func]   File "/path/to/lib/python3.6/site-packages/distributed/utils.py",
> line 1518, in __getitem__
>     value = super().__getitem__(key)   File "/path/to/lib/python3.6/collections/__init__.py",
> line 991, in __getitem__
>     raise KeyError(key) KeyError: <function interpolate_to_particles at 0x1228ce510>

从dask控制面板访问的工作日志不提供任何信息。事实上，我看不到工人除了开动外，还做了甚麽。

有没有关于可能发生的事情的想法，或者我可以用来进一步调试的建议工具？谢谢!

Answer 1

根据您的注释，您的函数听起来不能很好地序列化。要测试这一点，您可以尝试在一个进程中对函数进行酸洗，然后在另一个进程中尝试取消对其进行酸洗。

>>> import pickle
>>> print(pickle.dumps(interpolate_to_particles))
b'some bytes printed out here'

然后在另一个过程中

>>> import pickle
>>> interpolate_to_particles = pickle.loads(b'the same bytes you had before')

如果这不起作用，你就会知道那是你的问题。我鼓励你查阅“如何确保ctype函数是可序列化的”或类似的东西，或者在Stack Overflow上问另一个范围更小的问题。