Cryptic error when package not installed on worker

I'm a newbie dask distributed user, using it via kubernetes on a pangeo-like environment. I just made the blunder of trying to use map_blocks with a function that required a package that was installed in the notebook image but not in the worker image. When I initially did this, I used a map function like:

```python
def mapf(block):
return allel.GenotypeArray.is_missing(block)

result = gtd.map_blocks(mapf, drop_axis=2, dtype=bool).sum()
result.compute()
```

... where the package ``allel`` (scikit-allel) is not installed on the workers, and I got an error message like this:

```
---------------------------------------------------------------------------
KilledWorker Traceback (most recent call last)
in ()
----> 1 result.compute()

/opt/conda/lib/python3.6/site-packages/dask/base.py in compute(self, **kwargs)
154 dask.base.compute
155 """
--> 156 (result,) = compute(self, traverse=False, **kwargs)
157 return result
158

/opt/conda/lib/python3.6/site-packages/dask/base.py in compute(*args, **kwargs)
398 keys = [x.__dask_keys__() for x in collections]
399 postcomputes = [x.__dask_postcompute__() for x in collections]
--> 400 results = schedule(dsk, keys, **kwargs)
401 return repack([f(r, *a) for r, (f, a) in zip(results, postcomputes)])
402

/opt/conda/lib/python3.6/site-packages/distributed/client.py in get(self, dsk, keys, restrictions, loose_restrictions, resources, sync, asynchronous, direct, retries, priority, fifo_timeout, **kwargs)
2157 try:
2158 results = self.gather(packed, asynchronous=asynchronous,
-> 2159 direct=direct)
2160 finally:
2161 for f in futures.values():

/opt/conda/lib/python3.6/site-packages/distributed/client.py in gather(self, futures, errors, maxsize, direct, asynchronous)
1560 return self.sync(self._gather, futures, errors=errors,
1561 direct=direct, local_worker=local_worker,
-> 1562 asynchronous=asynchronous)
1563
1564 @gen.coroutine

/opt/conda/lib/python3.6/site-packages/distributed/client.py in sync(self, func, *args, **kwargs)
650 return future
651 else:
--> 652 return sync(self.loop, func, *args, **kwargs)
653
654 def __repr__(self):

/opt/conda/lib/python3.6/site-packages/distributed/utils.py in sync(loop, func, *args, **kwargs)
273 e.wait(10)
274 if error[0]:
--> 275 six.reraise(*error[0])
276 else:
277 return result[0]

/opt/conda/lib/python3.6/site-packages/six.py in reraise(tp, value, tb)
691 if value.__traceback__ is not tb:
692 raise value.with_traceback(tb)
--> 693 raise value
694 finally:
695 value = None

/opt/conda/lib/python3.6/site-packages/distributed/utils.py in f()
258 yield gen.moment
259 thread_state.asynchronous = True
--> 260 result[0] = yield make_coro()
261 except Exception as exc:
262 error[0] = sys.exc_info()

/opt/conda/lib/python3.6/site-packages/tornado/gen.py in run(self)
1097
1098 try:
-> 1099 value = future.result()
1100 except Exception:
1101 self.had_exception = True

/opt/conda/lib/python3.6/site-packages/tornado/gen.py in run(self)
1105 if exc_info is not None:
1106 try:
-> 1107 yielded = self.gen.throw(*exc_info)
1108 finally:
1109 # Break up a reference to itself

/opt/conda/lib/python3.6/site-packages/distributed/client.py in _gather(self, futures, errors, direct, local_worker)
1437 six.reraise(type(exception),
1438 exception,
-> 1439 traceback)
1440 if errors == 'skip':
1441 bad_keys.add(key)

/opt/conda/lib/python3.6/site-packages/six.py in reraise(tp, value, tb)
691 if value.__traceback__ is not tb:
692 raise value.with_traceback(tb)
--> 693 raise value
694 finally:
695 value = None

KilledWorker: ("('array-1647bba9096c6fe3f2cd4404e71e1839', 60, 0, 0)", 'tcp://10.8.87.4:36215')
```

I modify the map function a little I get a different error highlighting the lack of the allel package:

```
def mapf(block):
import allel
return allel.GenotypeArray.is_missing(block)

result = gtd.map_blocks(mapf, drop_axis=2, dtype=bool).sum()
result.compute()
```

...gives:

```
---------------------------------------------------------------------------
ModuleNotFoundError Traceback (most recent call last)
in ()
----> 1 result.compute()

/opt/conda/lib/python3.6/site-packages/dask/base.py in compute(self, **kwargs)
154 dask.base.compute
155 """
--> 156 (result,) = compute(self, traverse=False, **kwargs)
157 return result
158

/opt/conda/lib/python3.6/site-packages/dask/base.py in compute(*args, **kwargs)
398 keys = [x.__dask_keys__() for x in collections]
399 postcomputes = [x.__dask_postcompute__() for x in collections]
--> 400 results = schedule(dsk, keys, **kwargs)
401 return repack([f(r, *a) for r, (f, a) in zip(results, postcomputes)])
402

/opt/conda/lib/python3.6/site-packages/distributed/client.py in get(self, dsk, keys, restrictions, loose_restrictions, resources, sync, asynchronous, direct, retries, priority, fifo_timeout, **kwargs)
2157 try:
2158 results = self.gather(packed, asynchronous=asynchronous,
-> 2159 direct=direct)
2160 finally:
2161 for f in futures.values():

/opt/conda/lib/python3.6/site-packages/distributed/client.py in gather(self, futures, errors, maxsize, direct, asynchronous)
1560 return self.sync(self._gather, futures, errors=errors,
1561 direct=direct, local_worker=local_worker,
-> 1562 asynchronous=asynchronous)
1563
1564 @gen.coroutine

/opt/conda/lib/python3.6/site-packages/distributed/client.py in sync(self, func, *args, **kwargs)
650 return future
651 else:
--> 652 return sync(self.loop, func, *args, **kwargs)
653
654 def __repr__(self):

/opt/conda/lib/python3.6/site-packages/distributed/utils.py in sync(loop, func, *args, **kwargs)
273 e.wait(10)
274 if error[0]:
--> 275 six.reraise(*error[0])
276 else:
277 return result[0]

/opt/conda/lib/python3.6/site-packages/six.py in reraise(tp, value, tb)
691 if value.__traceback__ is not tb:
692 raise value.with_traceback(tb)
--> 693 raise value
694 finally:
695 value = None

/opt/conda/lib/python3.6/site-packages/distributed/utils.py in f()
258 yield gen.moment
259 thread_state.asynchronous = True
--> 260 result[0] = yield make_coro()
261 except Exception as exc:
262 error[0] = sys.exc_info()

/opt/conda/lib/python3.6/site-packages/tornado/gen.py in run(self)
1097
1098 try:
-> 1099 value = future.result()
1100 except Exception:
1101 self.had_exception = True

/opt/conda/lib/python3.6/site-packages/tornado/gen.py in run(self)
1105 if exc_info is not None:
1106 try:
-> 1107 yielded = self.gen.throw(*exc_info)
1108 finally:
1109 # Break up a reference to itself

/opt/conda/lib/python3.6/site-packages/distributed/client.py in _gather(self, futures, errors, direct, local_worker)
1437 six.reraise(type(exception),
1438 exception,
-> 1439 traceback)
1440 if errors == 'skip':
1441 bad_keys.add(key)

/opt/conda/lib/python3.6/site-packages/six.py in reraise(tp, value, tb)
690 value = tp()
691 if value.__traceback__ is not tb:
--> 692 raise value.with_traceback(tb)
693 raise value
694 finally:

in mapf()
1 def mapf(block):
----> 2 import allel
3 return allel.GenotypeArray.is_missing(block)

ModuleNotFoundError: No module named 'allel'
```

...so I *think* the lack of allel is also the cause for the original KilledWorker errors, but can't be completely sure. In any case, I realise this is my blunder, and I also realise there is a way to specify additional packages to install on the workers, but thought I would post this in case there was any way to improve error messages.

[Full reproducible example is here](https://gist.github.com/alimanfoo/7fad9309c2e88b98589b09a7415bb4a8#file-initial-tests-ipynb), can be run on pangeo.