can't pickle local object when doing a to_hdf5 when using dask.distributed

I'm playing around with the to_hdf5 command following some steps shown here:
http://dask.pydata.org/en/latest/array-creation.html

When I try to save the dask array to hdf5 file I got the following error:

```
distributed.protocol.pickle - INFO - Failed to serialize (.store at 0x7f72d2d1cc80>, (, , [(, 0, 3, 1, 3, dtype('int64'))], {'function': , 'other': [(1, 2)]}), (slice(0, 3, None),), , None)
Traceback (most recent call last):
File "/home/tangy/anaconda3/envs/ipykernel_py3/lib/python3.6/site-packages/distributed/protocol/pickle.py", line 41, in dumps
result = pickle.dumps(x, protocol=pickle.HIGHEST_PROTOCOL)
AttributeError: Can't pickle local object 'insert_to_ooc..store'

During handling of the above exception, another exception occurred:

Traceback (most recent call last):
File "/home/tangy/anaconda3/envs/ipykernel_py3/lib/python3.6/site-packages/distributed/protocol/pickle.py", line 54, in dumps
return cloudpickle.dumps(x, protocol=pickle.HIGHEST_PROTOCOL)
File "/home/tangy/anaconda3/envs/ipykernel_py3/lib/python3.6/site-packages/cloudpickle/cloudpickle.py", line 706, in dumps
cp.dump(obj)
File "/home/tangy/anaconda3/envs/ipykernel_py3/lib/python3.6/site-packages/cloudpickle/cloudpickle.py", line 146, in dump
return Pickler.dump(self, obj)
File "/home/tangy/anaconda3/envs/ipykernel_py3/lib/python3.6/pickle.py", line 409, in dump
self.save(obj)
File "/home/tangy/anaconda3/envs/ipykernel_py3/lib/python3.6/pickle.py", line 476, in save
f(self, obj) # Call unbound method with explicit self
File "/home/tangy/anaconda3/envs/ipykernel_py3/lib/python3.6/pickle.py", line 751, in save_tuple
save(element)
File "/home/tangy/anaconda3/envs/ipykernel_py3/lib/python3.6/pickle.py", line 476, in save
f(self, obj) # Call unbound method with explicit self
File "/home/tangy/anaconda3/envs/ipykernel_py3/lib/python3.6/site-packages/cloudpickle/cloudpickle.py", line 270, in save_function
self.save_function_tuple(obj)
File "/home/tangy/anaconda3/envs/ipykernel_py3/lib/python3.6/site-packages/cloudpickle/cloudpickle.py", line 312, in save_function_tuple
save((code, closure, base_globals))
File "/home/tangy/anaconda3/envs/ipykernel_py3/lib/python3.6/pickle.py", line 476, in save
f(self, obj) # Call unbound method with explicit self
File "/home/tangy/anaconda3/envs/ipykernel_py3/lib/python3.6/pickle.py", line 736, in save_tuple
save(element)
File "/home/tangy/anaconda3/envs/ipykernel_py3/lib/python3.6/pickle.py", line 476, in save
f(self, obj) # Call unbound method with explicit self
File "/home/tangy/anaconda3/envs/ipykernel_py3/lib/python3.6/pickle.py", line 781, in save_list
self._batch_appends(obj)
File "/home/tangy/anaconda3/envs/ipykernel_py3/lib/python3.6/pickle.py", line 808, in _batch_appends
save(tmp[0])
File "/home/tangy/anaconda3/envs/ipykernel_py3/lib/python3.6/pickle.py", line 521, in save
self.save_reduce(obj=obj, *rv)
File "/home/tangy/anaconda3/envs/ipykernel_py3/lib/python3.6/site-packages/cloudpickle/cloudpickle.py", line 604, in save_reduce
save(state)
File "/home/tangy/anaconda3/envs/ipykernel_py3/lib/python3.6/pickle.py", line 476, in save
f(self, obj) # Call unbound method with explicit self
File "/home/tangy/anaconda3/envs/ipykernel_py3/lib/python3.6/pickle.py", line 821, in save_dict
self._batch_setitems(obj.items())
File "/home/tangy/anaconda3/envs/ipykernel_py3/lib/python3.6/pickle.py", line 847, in _batch_setitems
save(v)
File "/home/tangy/anaconda3/envs/ipykernel_py3/lib/python3.6/pickle.py", line 496, in save
rv = reduce(self.proto)
TypeError: can't pickle h5py.h5d.DatasetID objects
distributed.protocol.core - CRITICAL - Failed to Serialize
Traceback (most recent call last):
File "/home/tangy/anaconda3/envs/ipykernel_py3/lib/python3.6/site-packages/distributed/protocol/pickle.py", line 41, in dumps
result = pickle.dumps(x, protocol=pickle.HIGHEST_PROTOCOL)
AttributeError: Can't pickle local object 'insert_to_ooc..store'

During handling of the above exception, another exception occurred:

Traceback (most recent call last):
File "/home/tangy/anaconda3/envs/ipykernel_py3/lib/python3.6/site-packages/distributed/protocol/core.py", line 47, in dumps
for key, value in data.items()
File "/home/tangy/anaconda3/envs/ipykernel_py3/lib/python3.6/site-packages/distributed/protocol/core.py", line 48, in
if type(value) is Serialize}
File "/home/tangy/anaconda3/envs/ipykernel_py3/lib/python3.6/site-packages/distributed/protocol/serialize.py", line 130, in serialize
header, frames = {}, [pickle.dumps(x)]
File "/home/tangy/anaconda3/envs/ipykernel_py3/lib/python3.6/site-packages/distributed/protocol/pickle.py", line 54, in dumps
return cloudpickle.dumps(x, protocol=pickle.HIGHEST_PROTOCOL)
File "/home/tangy/anaconda3/envs/ipykernel_py3/lib/python3.6/site-packages/cloudpickle/cloudpickle.py", line 706, in dumps
cp.dump(obj)
File "/home/tangy/anaconda3/envs/ipykernel_py3/lib/python3.6/site-packages/cloudpickle/cloudpickle.py", line 146, in dump
return Pickler.dump(self, obj)
File "/home/tangy/anaconda3/envs/ipykernel_py3/lib/python3.6/pickle.py", line 409, in dump
self.save(obj)
File "/home/tangy/anaconda3/envs/ipykernel_py3/lib/python3.6/pickle.py", line 476, in save
f(self, obj) # Call unbound method with explicit self
File "/home/tangy/anaconda3/envs/ipykernel_py3/lib/python3.6/pickle.py", line 751, in save_tuple
save(element)
File "/home/tangy/anaconda3/envs/ipykernel_py3/lib/python3.6/pickle.py", line 476, in save
f(self, obj) # Call unbound method with explicit self
File "/home/tangy/anaconda3/envs/ipykernel_py3/lib/python3.6/site-packages/cloudpickle/cloudpickle.py", line 270, in save_function
self.save_function_tuple(obj)
File "/home/tangy/anaconda3/envs/ipykernel_py3/lib/python3.6/site-packages/cloudpickle/cloudpickle.py", line 312, in save_function_tuple
save((code, closure, base_globals))
File "/home/tangy/anaconda3/envs/ipykernel_py3/lib/python3.6/pickle.py", line 476, in save
f(self, obj) # Call unbound method with explicit self
File "/home/tangy/anaconda3/envs/ipykernel_py3/lib/python3.6/pickle.py", line 736, in save_tuple
save(element)
File "/home/tangy/anaconda3/envs/ipykernel_py3/lib/python3.6/pickle.py", line 476, in save
f(self, obj) # Call unbound method with explicit self
File "/home/tangy/anaconda3/envs/ipykernel_py3/lib/python3.6/pickle.py", line 781, in save_list
self._batch_appends(obj)
File "/home/tangy/anaconda3/envs/ipykernel_py3/lib/python3.6/pickle.py", line 808, in _batch_appends
save(tmp[0])
File "/home/tangy/anaconda3/envs/ipykernel_py3/lib/python3.6/pickle.py", line 521, in save
self.save_reduce(obj=obj, *rv)
File "/home/tangy/anaconda3/envs/ipykernel_py3/lib/python3.6/site-packages/cloudpickle/cloudpickle.py", line 604, in save_reduce
save(state)
File "/home/tangy/anaconda3/envs/ipykernel_py3/lib/python3.6/pickle.py", line 476, in save
f(self, obj) # Call unbound method with explicit self
File "/home/tangy/anaconda3/envs/ipykernel_py3/lib/python3.6/pickle.py", line 821, in save_dict
self._batch_setitems(obj.items())
File "/home/tangy/anaconda3/envs/ipykernel_py3/lib/python3.6/pickle.py", line 847, in _batch_setitems
save(v)
File "/home/tangy/anaconda3/envs/ipykernel_py3/lib/python3.6/pickle.py", line 496, in save
rv = reduce(self.proto)
TypeError: can't pickle h5py.h5d.DatasetID objects
distributed.comm.utils - INFO - Unserializable Message: [{'op': 'update-graph', 'tasks': {"('arange-pow-store-pow-962eaabdc778ca6372a23fde28e7916c', 0)": .store at 0x7f72d2d1cc80>, (, , [(, 0, 3, 1, 3, dtype('int64'))], {'function': , 'other': [(1, 2)]}), (slice(0, 3, None),), , None)>, "('store-pow-962eaabdc778ca6372a23fde28e7916c', 0)": , "('arange-pow-store-pow-962eaabdc778ca6372a23fde28e7916c', 1)": .store at 0x7f72d2d1cc80>, (, , [(, 3, 6, 1, 3, dtype('int64'))], {'function': , 'other': [(1, 2)]}), (slice(3, 6, None),), , None)>, "('store-pow-962eaabdc778ca6372a23fde28e7916c', 1)": }, 'dependencies': {"('arange-pow-store-pow-962eaabdc778ca6372a23fde28e7916c', 0)": [], "('store-pow-962eaabdc778ca6372a23fde28e7916c', 0)": ["('arange-pow-store-pow-962eaabdc778ca6372a23fde28e7916c', 0)"], "('arange-pow-store-pow-962eaabdc778ca6372a23fde28e7916c', 1)": [], "('store-pow-962eaabdc778ca6372a23fde28e7916c', 1)": ["('arange-pow-store-pow-962eaabdc778ca6372a23fde28e7916c', 1)"]}, 'keys': ["('store-pow-962eaabdc778ca6372a23fde28e7916c', 1)", "('store-pow-962eaabdc778ca6372a23fde28e7916c', 0)"], 'restrictions': {}, 'loose_restrictions': None, 'priority': {"('store-pow-962eaabdc778ca6372a23fde28e7916c', 0)": 0, "('arange-pow-store-pow-962eaabdc778ca6372a23fde28e7916c', 0)": 1, "('store-pow-962eaabdc778ca6372a23fde28e7916c', 1)": 2, "('arange-pow-store-pow-962eaabdc778ca6372a23fde28e7916c', 1)": 3}, 'resources': None}]
distributed.comm.utils - ERROR - can't pickle h5py.h5d.DatasetID objects
Traceback (most recent call last):
File "/home/tangy/anaconda3/envs/ipykernel_py3/lib/python3.6/site-packages/distributed/protocol/pickle.py", line 41, in dumps
result = pickle.dumps(x, protocol=pickle.HIGHEST_PROTOCOL)
AttributeError: Can't pickle local object 'insert_to_ooc..store'

During handling of the above exception, another exception occurred:

Traceback (most recent call last):
File "/home/tangy/anaconda3/envs/ipykernel_py3/lib/python3.6/site-packages/distributed/comm/utils.py", line 16, in to_frames
return list(protocol.dumps(msg))
File "/home/tangy/anaconda3/envs/ipykernel_py3/lib/python3.6/site-packages/distributed/protocol/core.py", line 47, in dumps
for key, value in data.items()
File "/home/tangy/anaconda3/envs/ipykernel_py3/lib/python3.6/site-packages/distributed/protocol/core.py", line 48, in
if type(value) is Serialize}
File "/home/tangy/anaconda3/envs/ipykernel_py3/lib/python3.6/site-packages/distributed/protocol/serialize.py", line 130, in serialize
header, frames = {}, [pickle.dumps(x)]
File "/home/tangy/anaconda3/envs/ipykernel_py3/lib/python3.6/site-packages/distributed/protocol/pickle.py", line 54, in dumps
return cloudpickle.dumps(x, protocol=pickle.HIGHEST_PROTOCOL)
File "/home/tangy/anaconda3/envs/ipykernel_py3/lib/python3.6/site-packages/cloudpickle/cloudpickle.py", line 706, in dumps
cp.dump(obj)
File "/home/tangy/anaconda3/envs/ipykernel_py3/lib/python3.6/site-packages/cloudpickle/cloudpickle.py", line 146, in dump
return Pickler.dump(self, obj)
File "/home/tangy/anaconda3/envs/ipykernel_py3/lib/python3.6/pickle.py", line 409, in dump
self.save(obj)
File "/home/tangy/anaconda3/envs/ipykernel_py3/lib/python3.6/pickle.py", line 476, in save
f(self, obj) # Call unbound method with explicit self
File "/home/tangy/anaconda3/envs/ipykernel_py3/lib/python3.6/pickle.py", line 751, in save_tuple
save(element)
File "/home/tangy/anaconda3/envs/ipykernel_py3/lib/python3.6/pickle.py", line 476, in save
f(self, obj) # Call unbound method with explicit self
File "/home/tangy/anaconda3/envs/ipykernel_py3/lib/python3.6/site-packages/cloudpickle/cloudpickle.py", line 270, in save_function
self.save_function_tuple(obj)
File "/home/tangy/anaconda3/envs/ipykernel_py3/lib/python3.6/site-packages/cloudpickle/cloudpickle.py", line 312, in save_function_tuple
save((code, closure, base_globals))
File "/home/tangy/anaconda3/envs/ipykernel_py3/lib/python3.6/pickle.py", line 476, in save
f(self, obj) # Call unbound method with explicit self
File "/home/tangy/anaconda3/envs/ipykernel_py3/lib/python3.6/pickle.py", line 736, in save_tuple
save(element)
File "/home/tangy/anaconda3/envs/ipykernel_py3/lib/python3.6/pickle.py", line 476, in save
f(self, obj) # Call unbound method with explicit self
File "/home/tangy/anaconda3/envs/ipykernel_py3/lib/python3.6/pickle.py", line 781, in save_list
self._batch_appends(obj)
File "/home/tangy/anaconda3/envs/ipykernel_py3/lib/python3.6/pickle.py", line 808, in _batch_appends
save(tmp[0])
File "/home/tangy/anaconda3/envs/ipykernel_py3/lib/python3.6/pickle.py", line 521, in save
self.save_reduce(obj=obj, *rv)
File "/home/tangy/anaconda3/envs/ipykernel_py3/lib/python3.6/site-packages/cloudpickle/cloudpickle.py", line 604, in save_reduce
save(state)
File "/home/tangy/anaconda3/envs/ipykernel_py3/lib/python3.6/pickle.py", line 476, in save
f(self, obj) # Call unbound method with explicit self
File "/home/tangy/anaconda3/envs/ipykernel_py3/lib/python3.6/pickle.py", line 821, in save_dict
self._batch_setitems(obj.items())
File "/home/tangy/anaconda3/envs/ipykernel_py3/lib/python3.6/pickle.py", line 847, in _batch_setitems
save(v)
File "/home/tangy/anaconda3/envs/ipykernel_py3/lib/python3.6/pickle.py", line 496, in save
rv = reduce(self.proto)
TypeError: can't pickle h5py.h5d.DatasetID objects
distributed.batched - ERROR - Error in batched write
Traceback (most recent call last):
File "/home/tangy/anaconda3/envs/ipykernel_py3/lib/python3.6/site-packages/distributed/protocol/pickle.py", line 41, in dumps
result = pickle.dumps(x, protocol=pickle.HIGHEST_PROTOCOL)
AttributeError: Can't pickle local object 'insert_to_ooc..store'

During handling of the above exception, another exception occurred:

Traceback (most recent call last):
File "/home/tangy/anaconda3/envs/ipykernel_py3/lib/python3.6/site-packages/distributed/batched.py", line 85, in _background_send
nbytes = yield self.comm.write(payload)
File "/home/tangy/anaconda3/envs/ipykernel_py3/lib/python3.6/site-packages/tornado/gen.py", line 1015, in run
value = future.result()
File "/home/tangy/anaconda3/envs/ipykernel_py3/lib/python3.6/site-packages/tornado/concurrent.py", line 237, in result
raise_exc_info(self._exc_info)
File "", line 3, in raise_exc_info
File "/home/tangy/anaconda3/envs/ipykernel_py3/lib/python3.6/site-packages/tornado/gen.py", line 270, in wrapper
result = func(*args, **kwargs)
File "/home/tangy/anaconda3/envs/ipykernel_py3/lib/python3.6/types.py", line 248, in wrapped
coro = func(*args, **kwargs)
File "/home/tangy/anaconda3/envs/ipykernel_py3/lib/python3.6/site-packages/distributed/comm/tcp.py", line 171, in write
frames = [ensure_bytes(f) for f in to_frames(msg)]
File "/home/tangy/anaconda3/envs/ipykernel_py3/lib/python3.6/site-packages/distributed/comm/utils.py", line 16, in to_frames
return list(protocol.dumps(msg))
File "/home/tangy/anaconda3/envs/ipykernel_py3/lib/python3.6/site-packages/distributed/protocol/core.py", line 47, in dumps
for key, value in data.items()
File "/home/tangy/anaconda3/envs/ipykernel_py3/lib/python3.6/site-packages/distributed/protocol/core.py", line 48, in
if type(value) is Serialize}
File "/home/tangy/anaconda3/envs/ipykernel_py3/lib/python3.6/site-packages/distributed/protocol/serialize.py", line 130, in serialize
header, frames = {}, [pickle.dumps(x)]
File "/home/tangy/anaconda3/envs/ipykernel_py3/lib/python3.6/site-packages/distributed/protocol/pickle.py", line 54, in dumps
return cloudpickle.dumps(x, protocol=pickle.HIGHEST_PROTOCOL)
File "/home/tangy/anaconda3/envs/ipykernel_py3/lib/python3.6/site-packages/cloudpickle/cloudpickle.py", line 706, in dumps
cp.dump(obj)
File "/home/tangy/anaconda3/envs/ipykernel_py3/lib/python3.6/site-packages/cloudpickle/cloudpickle.py", line 146, in dump
return Pickler.dump(self, obj)
File "/home/tangy/anaconda3/envs/ipykernel_py3/lib/python3.6/pickle.py", line 409, in dump
self.save(obj)
File "/home/tangy/anaconda3/envs/ipykernel_py3/lib/python3.6/pickle.py", line 476, in save
f(self, obj) # Call unbound method with explicit self
File "/home/tangy/anaconda3/envs/ipykernel_py3/lib/python3.6/pickle.py", line 751, in save_tuple
save(element)
File "/home/tangy/anaconda3/envs/ipykernel_py3/lib/python3.6/pickle.py", line 476, in save
f(self, obj) # Call unbound method with explicit self
File "/home/tangy/anaconda3/envs/ipykernel_py3/lib/python3.6/site-packages/cloudpickle/cloudpickle.py", line 270, in save_function
self.save_function_tuple(obj)
File "/home/tangy/anaconda3/envs/ipykernel_py3/lib/python3.6/site-packages/cloudpickle/cloudpickle.py", line 312, in save_function_tuple
save((code, closure, base_globals))
File "/home/tangy/anaconda3/envs/ipykernel_py3/lib/python3.6/pickle.py", line 476, in save
f(self, obj) # Call unbound method with explicit self
File "/home/tangy/anaconda3/envs/ipykernel_py3/lib/python3.6/pickle.py", line 736, in save_tuple
save(element)
File "/home/tangy/anaconda3/envs/ipykernel_py3/lib/python3.6/pickle.py", line 476, in save
f(self, obj) # Call unbound method with explicit self
File "/home/tangy/anaconda3/envs/ipykernel_py3/lib/python3.6/pickle.py", line 781, in save_list
self._batch_appends(obj)
File "/home/tangy/anaconda3/envs/ipykernel_py3/lib/python3.6/pickle.py", line 808, in _batch_appends
save(tmp[0])
File "/home/tangy/anaconda3/envs/ipykernel_py3/lib/python3.6/pickle.py", line 521, in save
self.save_reduce(obj=obj, *rv)
File "/home/tangy/anaconda3/envs/ipykernel_py3/lib/python3.6/site-packages/cloudpickle/cloudpickle.py", line 604, in save_reduce
save(state)
File "/home/tangy/anaconda3/envs/ipykernel_py3/lib/python3.6/pickle.py", line 476, in save
f(self, obj) # Call unbound method with explicit self
File "/home/tangy/anaconda3/envs/ipykernel_py3/lib/python3.6/pickle.py", line 821, in save_dict
self._batch_setitems(obj.items())
File "/home/tangy/anaconda3/envs/ipykernel_py3/lib/python3.6/pickle.py", line 847, in _batch_setitems
save(v)
File "/home/tangy/anaconda3/envs/ipykernel_py3/lib/python3.6/pickle.py", line 496, in save
rv = reduce(self.proto)
TypeError: can't pickle h5py.h5d.DatasetID objects
```

Here is the code:
```
import dask.array as da
import numpy as np
import distributed;
client = distributed.Client()
x = da.arange(6,chunks=3)
y = x**2
np.array(y)
y.compute()
da.to_hdf5('myfile.hdf5', '/y', y)
```
I am running a conda environment with Python 3.6 installed. All schedulers, workers, and clients are running within this environment.