Dask Distributed appears to not be passing the requested operation during compute()

Question

In the below operation (adapted from the Dask DataFrame API docs), if I don't attach to a scheduler (leave the line assigning the client variable commented out), the operation completes successfully as expected.

from dask.distributed import Client
import dask.dataframe as dd
import pandas as pd

connection_loc = 'foobar.net:8786'
# client = Client(connection_loc)

df = pd.DataFrame({'x': [1, 2, 3, 4, 5], 'y': [1., 2., 3., 4., 5.]})
ddf = dd.from_pandas(df, npartitions=2)
foo = ddf.map_overlap(lambda df: df.rolling(2).sum(), 2, 0).compute()

The moment that same line is uncommented and a client connection is assigned, the following error occurs: TypeError: unorderable types: list() >= int() (see full traceback for more).

Examining the traceback, I can see that the bytestring it is trying to deserialize is not what I would expect it should be trying to deserialize (see first line in full traceback distributed.protocol.pickle - INFO - Failed to deserialize).

I've completely stopped and restarted the remote containers running both the worker and the scheduler to no avail. I've also used client.restart() with no luck. Any idea why this other task is being passed to the worker and throwing this error? Any solution to get Dask to stop doing this?

Full traceback:

dask_worker_1     | distributed.protocol.pickle - INFO - Failed to deserialize b"\x80\x04\x95+\x01\x00\x00\x00\x00\x00\x00(\x8c\x17cloudpickle.cloudpickle\x94\x8c\x0e_fill_function\x94\x93\x94(h\x00\x8c\x0f_make_skel_func\x94\x93\x94h\x00\x8c\r_builtin_type\x94\x93\x94\x8c\x08CodeType\x94\x85\x94R\x94(K\x01K\x00K\x01K\x02KCC\x0e|\x00j\x00d\x01\x83\x01j\x01\x83\x00S\x00\x94NK\x02\x86\x94\x8c\x07rolling\x94\x8c\x03sum\x94\x86\x94\x8c\x02df\x94\x85\x94\x8c\x1fdask_method/dask_dist_matrix.py\x94\x8c\x08<lambda>\x94K\rC\x00\x94))t\x94R\x94]\x94}\x94\x87\x94R\x94}\x94N}\x94tRN\x8c3('from_pandas-ddc065084280667dd51853b144bdd4e8', 0)\x94NK\x02K\x00)}\x94t\x94."
dask_worker_1     | Traceback (most recent call last):
dask_worker_1     |   File "/usr/local/lib/python3.5/site-packages/distributed/protocol/pickle.py", line 59, in loads
dask_worker_1     |     return pickle.loads(x)
dask_worker_1     |   File "/usr/local/lib/python3.5/site-packages/cloudpickle/cloudpickle.py", line 935, in _make_skel_func
dask_worker_1     |     if cell_count >= 0 else
dask_worker_1     | TypeError: unorderable types: list() >= int()
dask_worker_1     | distributed.worker - WARNING - Could not deserialize task
dask_worker_1     | Traceback (most recent call last):
dask_worker_1     |   File "/usr/local/lib/python3.5/site-packages/distributed/worker.py", line 1113, in add_task
dask_worker_1     |     self.tasks[key] = _deserialize(function, args, kwargs, task)
dask_worker_1     |   File "/usr/local/lib/python3.5/site-packages/distributed/worker.py", line 573, in _deserialize
dask_worker_1     |     args = pickle.loads(args)
dask_worker_1     |   File "/usr/local/lib/python3.5/site-packages/distributed/protocol/pickle.py", line 59, in loads
dask_worker_1     |     return pickle.loads(x)
dask_worker_1     |   File "/usr/local/lib/python3.5/site-packages/cloudpickle/cloudpickle.py", line 935, in _make_skel_func
dask_worker_1     |     if cell_count >= 0 else
dask_worker_1     | TypeError: unorderable types: list() >= int()

Dask: 0.15.0 Distributed: 1.17.1 OS: Ubuntu 16.04.2 LTS

Answer 1

I suspect that you have a mismatch in your cloudpickle versions between workers and clients. You'll have to ensure that all of your workers and clients have the same software setup. You can try the following command to help:

client.get_versions(check=True)

I don't think that this includes cloudpickle in dask.distributed version 1.17.1 but should in all subsequent versions. (it works now in master)

Answer 2

As the other answer mentions, this is almost certainly a mismatch in software versions. I had the same problem.

I did several things to get it all working again. I was using dask_ec2, so I'll include those changes here, but I don't know how you're setting up your cluster.

First, as I was using ubuntu 16.04 locally, I imagined that it was more likely to have the same libraries etc if the distributed servers had the same version, but this had an issue (see https://github.com/dask/dask-ec2/issues/98 ). Summary: I modified dask_ec2/salt.py, setting it to download cherrypy==3.2.3 in the __install_salt_rest_api method (see the linked issue for more detail).

Second, I set dask_ec2 to use the newer versions of Anaconda. In dask_ec2/formulas/salt/conda/settings.sls, change the download_url lines to:

{% set download_url = 'https://repo.continuum.io/archive/Anaconda2-5.0.1-Linux-x86_64.sh' %}

{% set download_url = 'https://repo.continuum.io/archive/Anaconda3-5.0.1-Linux-x86_64.sh' %}

Third, I ran an update on my own computer, to ensure my own libraries were up to date:

E.g. from: Upgrading all packages with pip

pip freeze --local | grep -v '^\-e' | cut -d = -f 1  | xargs -n1 pip install -U

and

conda update --all

I finally restarted the whole lot, and it worked fine.