Inconsistent unpickling errors in different modes between REPL and CLI

The following works in both cluster and local mode from the command line, but in the REPL it succeeds in local model but fails in cluster mode. I am guessing this is just an unpickling PYTHONPATH error but am wondering what are the typical paths to debugging and working around it?

In general, this comes up while trying to submit jobs to a cluster to iterate on some development modules. I think some sort of wrapper to set PYTHONPATH dynamically before the unpickle will solve this.

example.py:

```
#!/usr/bin/env python
import sys
import dask
from distributed import Client
# from dask.distributed import Client
import dask.bag as db

def print_and_return_message(msg):
print("msg:", msg)
return "print: [{}]".format(msg)

def submit(client):
r = client.submit(print_and_return_message, 'what')
print("r", r.result())

def test():
b = db.from_sequence([1, 2, 3, 4, 5, 6, 7, 8, 9, 10])
return b.map(print_and_return_message)

def cluster():
client = Client('127.0.0.1:8786')
submit(client)

def local():
# dask.set_options(get=dask.local.get_sync)
# client = Client()
# or
client = Client(processes=False)
submit(client)

if __name__ == '__main__':
if len(sys.argv) > 1 and sys.argv[1] == 'local':
print("running local mode")
local()
else:
cluster()
print("running cluster mode")
```

CLI

```
~/projects/notebooks/dask $ python example.py cluster
r print: [what]
running cluster mode
~/projects/notebooks/dask $ python example.py local
running local mode
msg: what
r print: [what]
```

REPL

```
In [1]: import example as e

In [2]: e.cluster()
---------------------------------------------------------------------------
ModuleNotFoundError Traceback (most recent call last)
in ()
----> 1 e.cluster()

~/projects/notebooks/dask/example.py in cluster()
20 def cluster():
21 client = Client('127.0.0.1:8786')
---> 22 submit(client)
23
24 def local():

~/projects/notebooks/dask/example.py in submit(client)
12 def submit(client):
13 r = client.submit(print_and_return_message, 'what')
---> 14 print("r", r.result())
15
16 def test():

~/anaconda3/envs/363/lib/python3.6/site-packages/distributed/client.py in result(self, timeout)
169 raiseit=False)
170 if self.status == 'error':
--> 171 six.reraise(*result)
172 elif self.status == 'cancelled':
173 raise result

~/anaconda3/envs/363/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:

~/anaconda3/envs/363/lib/python3.6/site-packages/distributed/protocol/pickle.py in loads()
57 def loads(x):
58 try:
---> 59 return pickle.loads(x)
60 except Exception:
61 logger.info("Failed to deserialize %s", x[:10000], exc_info=True)

ModuleNotFoundError: No module named 'example'

In [3]: e.local()
msg: what
r print: [what]
```