Replicate a dataset with dask to all workers
<p>I am using dask with distributed scheduler. I am trying to replicate a dataset read through csv on s3 to all worker nodes. Example:</p>

<pre><code>from distributed import Executor
import dask.dataframe as dd

e= Executor('127.0.0.1:8786',set_as_default=True)
df = dd.read_csv('s3://bucket/file.csv', blocksize=None) 
df = e.persist(df)
e.replicate(df)

distributed.utils - ERROR - unhashable type: 'list'
Traceback (most recent call last):
  File "/root/.miniconda/envs/dask_env/lib/python3.5/site-packages/distributed/utils.py", line 102, in f
    result[0] = yield gen.maybe_future(func(*args, **kwargs))
  File "/root/.miniconda/envs/dask_env/lib/python3.5/site-packages/tornado/gen.py", line 1015, in run
    value = future.result()
  File "/root/.miniconda/envs/dask_env/lib/python3.5/site-packages/tornado/concurrent.py", line 237, in result
    raise_exc_info(self._exc_info)
  File "&lt;string&gt;", line 3, in raise_exc_info
  File "/root/.miniconda/envs/dask_env/lib/python3.5/site-packages/tornado/gen.py", line 1021, in run
    yielded = self.gen.throw(*exc_info)
  File "/root/.miniconda/envs/dask_env/lib/python3.5/site-packages/distributed/executor.py", line 1347, in _replicate
    branching_factor=branching_factor)
  File "/root/.miniconda/envs/dask_env/lib/python3.5/site-packages/tornado/gen.py", line 1015, in run
    value = future.result()
  File "/root/.miniconda/envs/dask_env/lib/python3.5/site-packages/tornado/concurrent.py", line 237, in result
    raise_exc_info(self._exc_info)
  File "&lt;string&gt;", line 3, in raise_exc_info
  File "/root/.miniconda/envs/dask_env/lib/python3.5/site-packages/tornado/gen.py", line 1021, in run
    yielded = self.gen.throw(*exc_info)
  File "/root/.miniconda/envs/dask_env/lib/python3.5/site-packages/distributed/core.py", line 444, in send_recv_from_rpc
    result = yield send_recv(stream=stream, op=key, **kwargs)
  File "/root/.miniconda/envs/dask_env/lib/python3.5/site-packages/tornado/gen.py", line 1015, in run
    value = future.result()
  File "/root/.miniconda/envs/dask_env/lib/python3.5/site-packages/tornado/concurrent.py", line 237, in result
    raise_exc_info(self._exc_info)
  File "&lt;string&gt;", line 3, in raise_exc_info
  File "/root/.miniconda/envs/dask_env/lib/python3.5/site-packages/tornado/gen.py", line 1024, in run
    yielded = self.gen.send(value)
  File "/root/.miniconda/envs/dask_env/lib/python3.5/site-packages/distributed/core.py", line 345, in send_recv
    six.reraise(*clean_exception(**response))
  File "/root/.miniconda/envs/dask_env/lib/python3.5/site-packages/six.py", line 685, in reraise
    raise value.with_traceback(tb)
  File "/root/.miniconda/envs/dask_env/lib/python3.5/site-packages/distributed/core.py", line 211, in handle_stream
    result = yield gen.maybe_future(handler(stream, **msg))
  File "/root/.miniconda/envs/dask_env/lib/python3.5/site-packages/tornado/gen.py", line 1015, in run
    value = future.result()
  File "/root/.miniconda/envs/dask_env/lib/python3.5/site-packages/tornado/concurrent.py", line 237, in result
    raise_exc_info(self._exc_info)
  File "&lt;string&gt;", line 3, in raise_exc_info
  File "/root/.miniconda/envs/dask_env/lib/python3.5/site-packages/tornado/gen.py", line 285, in wrapper
    yielded = next(result)
  File "/root/.miniconda/envs/dask_env/lib/python3.5/site-packages/distributed/scheduler.py", line 1324, in replicate
    keys = set(keys)
TypeError: unhashable type: 'list'
</code></pre>

<p>Is this the correct way to replicate a dataframe? It appears that <code>e.persist(df)</code> returned object does not work with <code>e.replicate</code> for some reason.</p>

<p>This was a bug and has been resolved in <a href="https://github.com/dask/distributed/pull/473" rel="nofollow">https://github.com/dask/distributed/pull/473</a></p>

python dask