Распределенный Dask Не удалось десериализовать с помощью numpy.arrays и sparse.matrices

Я получаю следующую ошибку несколько раз для разных задач на графике (изменения между выполнениями). Возможно, когда определенные задачи возвращают 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-distribution, cloudpickle)