title: dask distributed memory error
body: <p>I got the following error on the scheduler while running Dask on a distributed job:</p>
  
  <pre><code>distributed.core - ERROR -
  Traceback (most recent call last):
    File "/usr/local/lib/python3.4/dist-packages/distributed/core.py", line 269, in write
      frames = protocol.dumps(msg)
    File "/usr/local/lib/python3.4/dist-packages/distributed/protocol.py", line 81, in dumps
      frames = dumps_msgpack(small)
    File "/usr/local/lib/python3.4/dist-packages/distributed/protocol.py", line 153, in dumps_msgpack
      payload = msgpack.dumps(msg, use_bin_type=True)
    File "/usr/local/lib/python3.4/dist-packages/msgpack/__init__.py", line 47, in packb
      return Packer(**kwargs).pack(o)
    File "msgpack/_packer.pyx", line 231, in msgpack._packer.Packer.pack (msgpack/_packer.cpp:231)
    File "msgpack/_packer.pyx", line 239, in msgpack._packer.Packer.pack (msgpack/_packer.cpp:239)
  MemoryError
  </code></pre>
  
  <p>Is this running out of memory on the scheduler or on one of the workers? Or both??</p>