dask.delayed KeyError 与分布式调度程序

Question

我有一个函数interpolate_to_particles 用c 编写并用ctypes 包装。我想使用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

鉴于您的 cmets，听起来您的函数不能很好地序列化。要对此进行测试，您可以尝试在一个进程中对函数进行酸洗，然后在另一个进程中尝试对其进行取消酸洗。

>>> 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')

如果这不起作用，那么您就会知道这是您的问题。我鼓励您查看“如何确保 ctypes 函数是可序列化的”或类似的内容，或者在 Stack Overflow 上问另一个范围较小的问题。