Dask 分布式无法使用 numpy.arrays 和 sparse.matrices 反序列化答案

【问题标题】：Dask distributed Failed to deserialize with numpy.arrays and sparse.matricesDask 分布式无法使用 numpy.arrays 和 sparse.matrices 反序列化
【发布时间】：2019-02-27 12:17:24
【问题描述】：

我在图表中的不同任务上多次收到以下错误（执行之间的更改）。可能当某些任务返回 numpy.arrays 和 scipy.sparse 矩阵时。

distributed.protocol.pickle - INFO - Failed to deserialize b'\x80\x04'
Traceback (most recent call last):
  File "/home/user/venv/lib/python3.5/site-packages/distributed/protocol/pickle.py", line 59, in loads
    return pickle.loads(x)
EOFError: Ran out of input
distributed.protocol.core - CRITICAL - Failed to deserialize
Traceback (most recent call last):
  File "/home/user/venv/lib/python3.5/site-packages/distributed/protocol/core.py", line 119, in loads
    value = _deserialize(head, fs)
  File "/home/user/venv/lib/python3.5/site-packages/distributed/protocol/serialize.py", line 158, in deserialize
    return f(header, frames)
  File "/home/user/venv/lib/python3.5/site-packages/distributed/protocol/serialize.py", line 20, in <lambda>
    deserializers = {None: lambda header, frames: pickle.loads(b''.join(frames))}
  File "/home/user/venv/lib/python3.5/site-packages/distributed/protocol/pickle.py", line 59, in loads
    return pickle.loads(x)
EOFError: Ran out of input
distributed.comm.utils - ERROR - truncated data stream (485 bytes): [b'', b"\x92\x83\xa6report\xc2\xa4keys\x91\xd9P('_avro_body-read-block-bag-from-delayed-67c7a9690149de9743ed970f873fa1d6', 283)\xa2op\xabdelete-data\x86\xa8priority\x93\x00\x01\xcc\xce\xa6nbytes\x81\xd9:('bag-from-delayed-67c7a9690149de9743ed970f873fa1d6', 283)\xce\x00 \x86p\xa8duration\xcb@\x18\x16m\x88xX\x00\xa7who_has\x81\xd9:('bag-from-delayed-67c7a9690149de9743ed970f873fa1d6', 283)\x91\xb5tcp://127.0.0.1:38623\xa2op\xaccompute-task\xa3key\xd9K('pluck-map-process_features_sparse-d94d304dc59efb780c39bfb0ca4df37f', 283)", b'\x83\xabbytestrings\x90\xa7headers\x81\x92\x01\xa4task\x83\xabcompression\x91\xc0\xa5count\x01\xa7lengths\x91\x02\xa4keys\x91\x92\x01\xa4task', b'\x80\x04']
distributed.worker - INFO - Connection to scheduler broken. Reregistering
distributed.worker - INFO - -------------------------------------------------
distributed.worker - INFO -         Registered to:       tcp://127.0.0.1:8786
distributed.worker - INFO - -------------------------------------------------

这始终是 EOFError: Ran out of input 错误，具有不同大小的缓冲区（有时小到几个字节），整个集群在一台机器上运行。

理想情况下，我希望找到实际问题的解决方案，但我们也将不胜感激调查问题和了解可能出现问题的方法。现在我有点卡住了，不知道如何解决手头的问题。

运行 client.get_versions(check=True) 完成时没有错误，并且在更新所有包（即 numpy、scipy、dask、dask-distributed、cloudpickle）后仍然存在

【问题讨论】：

我不知道。我建议尝试生成mcve
会的。我应该打开一个 github 问题还是用最少的复制来编辑问题？此外，将, b'\x80\x04'] 作为数据流列表中的第三项是否合理？
随你喜欢。对于您的b'\x80\x04'] 问题，我没有答案。
在 ray-project 的一个 issue[1] 中也提到了 b'\x80\x04' 标头。他们将其追溯到cloudpickle 中的一个错误，因此升级cloudpickle 可能会解决问题。 [1]：github.com/ray-project/ray/issues/2685#issuecomment-423182347

标签： dask dask-distributed

【解决方案1】：

cloudpickle 项目（dask 使用）最近已修补以修复可能导致此错误的问题。

此评论中解释了一些细节： https://github.com/ray-project/ray/issues/2685#issuecomment-423182347

...更多详细信息可以在cloudpickle github repo 中的相关问题/PR 中找到。

FWIW，我今天遇到了这个错误（包括b'\x80\x04' 部分），将cloudpickle 更新为0.8.0 似乎已经修复了它。

【讨论】：