Array > 2GB hitting `msgpack` limit

[gforsyth]

I know that there's supposed to be alternate protocols used for larger arrays but I'm not sure what needs to done to use them (or if they don't play nice with map_blocks?

MRE:

import numpy
import dask.array as da
from distributed import Client

client = Client('127.0.0.1:8786')
def increment_by_one(my_array):
    return my_array + 1

data = numpy.random.random(300000000)
chunks = (10000,)
data = da.from_array(data, chunks=chunks)
output = da.map_blocks(increment_by_one, data)
output.compute()

Traceback:

distributed.utils - ERROR - 2400000161 exceeds max_bin_len(2147483647)
Traceback (most recent call last):
  File "/home/gil/anaconda/envs/dasknumbagpu/lib/python3.5/site-packages/distributed/utils.py", line 207, in log_errors
    yield
  File "/home/gil/anaconda/envs/dasknumbagpu/lib/python3.5/site-packages/distributed/client.py", line 460, in _handle_report
    six.reraise(*clean_exception(**msg))
  File "/home/gil/anaconda/envs/dasknumbagpu/lib/python3.5/site-packages/six.py", line 685, in reraise
    raise value.with_traceback(tb)
  File "/home/gil/anaconda/envs/dasknumbagpu/lib/python3.5/site-packages/tornado/concurrent.py", line 237, in result
    raise_exc_info(self._exc_info)
  File "<string>", line 3, in raise_exc_info
  File "/home/gil/anaconda/envs/dasknumbagpu/lib/python3.5/site-packages/tornado/gen.py", line 1024, in run
    yielded = self.gen.send(value)
  File "/home/gil/anaconda/envs/dasknumbagpu/lib/python3.5/site-packages/distributed/core.py", line 258, in read
    msg = protocol.loads(frames)
  File "/home/gil/anaconda/envs/dasknumbagpu/lib/python3.5/site-packages/distributed/protocol.py", line 152, in loads
    msg = loads_msgpack(small_header, small_payload)
  File "/home/gil/anaconda/envs/dasknumbagpu/lib/python3.5/site-packages/distributed/protocol.py", line 256, in loads_msgpack
    return msgpack.loads(payload, encoding='utf8')
  File "pandas/msgpack/_unpacker.pyx", line 138, in pandas.msgpack._unpacker.unpackb (pandas/msgpack/_unpacker.cpp:2059)
ValueError: 2400000161 exceeds max_bin_len(2147483647)

[mrocklin]

I can reproduce the issue with the following slightly simpler example

In [1]: import numpy
   ...: import dask.array as da
   ...: from distributed import Client
   ...: 
   ...: client = Client()
   ...: 
   ...: data = numpy.random.random(300000000)
   ...: chunks = (10000,)
   ...: data = da.from_array(data, chunks=chunks)
   ...: data = client.persist(data)

So there are a couple things going on here:

Protocol error

Apparently some large bytestrings are able to slip through the protocol (which usually cleans them out and sends them separately) and into msgpack, which as you've noted, can't handle them. This needs to be fixed internally

User behaviors

Generally we should avoid putting large amounts of data in the dask graph. Now of course, it's note entirely clear when you're doing this. This happens whenever you use a dask.array function that consumes a large numpy array and gives you back a dask.array. If you look at the .dask property afterwards you'll notice that there are numpy arrays within it. This .dask attribute gets sent up to the scheduler, which we generally want to keep fairly light, so our goal is to make it so that data stays out of the .dask graph, either by using loading functions (like da.random.random or da.from_delayed) or futures, both of which are shown below. This is important for performance even after we fix the protocol error above.

It's best to load data direction on the workers if possible. This can be done with dask.array functions like da.random.random, or with client.submit/map or with dask.delayed. This doc page might be interesting: http://distributed.readthedocs.io/en/latest/memory.html

x = da.random.random(size=..., chunks=...)

or

data = e.submit(np.random.random, size)   # produces a single future

or

data = np.random.random(...)
future = client.scatter(data)  # produces a single future

From this future you can construct a dask.array of a single chunk and then rechunk it. Here is a comprehensive example using scatter:

In [1]: from distributed import Client

In [2]: client = Client()

In [3]: import numpy as np

In [4]: data = np.random.random(10000000)

In [5]: data.nbytes / 1e6
Out[5]: 80.0

In [6]: %time [future] = client.scatter([data])
CPU times: user 1.08 s, sys: 132 ms, total: 1.21 s
Wall time: 1.25 s

In [7]: import dask.array as da

In [8]: from dask import delayed

In [22]: x = da.from_delayed(delayed(future), shape=data.shape, dtype=data.dtype)

In [23]: x
Out[23]: dask.array<from-va..., shape=(10000000,), dtype=float64, chunksize=(10000000,)>

In [24]: x = x.rechunk((30000,))

In [25]: x
Out[25]: dask.array<rechunk..., shape=(10000000,), dtype=float64, chunksize=(30000,)>

In [26]: x = client.persist(x)

In [28]: {w: len(keys) for w, keys in client.has_what().items()}  # but wait!  all data on the same worker
Out[28]: 
{'25.231.111.255:35376': 0,
 '25.231.111.255:38110': 335,
 '25.231.111.255:42691': 0,
 '25.231.111.255:45072': 0}

In [29]: client.rebalance(x)

In [30]: {w: len(keys) for w, keys in client.has_what().items()}
Out[30]: 
{'25.231.111.255:35376': 82,
 '25.231.111.255:38110': 85,
 '25.231.111.255:42691': 84,
 '25.231.111.255:45072': 84}

However this may not be your actual problem. I've just optimized the fail case you've provided. This may not be the best approach to what you actually want to do, so it's best to take this all with some salt.

[mrocklin]

OK, so the protocol is failing here because its a single message of very many small parts.

None of the chunks are individually large enough to warrant being pulled out into a separate frame and yet when taken all together the message is long enough to break msgpack. I didn't realize that the 2GB msgpack limit wasn't per-value in the message but rather a limit over the entire message.

This could become problematic in other cases as well. I don't currently see a simple fix other than to better inform users on how to avoid these situations. We would need to become better at picking apart arbitrary messages and then merging them back together downstream. This isn't trivial to do generally over all possible msgpack messages.

[mrocklin]

cc @jcrist in case he has ideas

[jcrist]

I've pushed a fix for this in #530. The issue was that we weren't extracting bytestrings from messages if the message is a list. This was originally done because often (but not always, as this issue has shown) messages that are lists are batches of small messages, and this was an attempt at efficiency. However, after profiling the scheduler the overhead of traversing both lists and dicts is negligible compared to the normal computational costs.

Just to clarify, the issue Matt brought up above is incorrect and doesn't occur. Msgpack doesn't care about the size of the total message, you just can't have bytestrings > 2 GB. The message being sent was composed of a list of many small bytestrings, and a single large bytestring (the original array), which wasn't being broken apart due to not traversing top-level dictionaries.

[RokoMijic]

I think I may be hitting this error.

dd.merge(left,right,on=keycols, how='outer')

yields:

---------------------------------------------------------------------------
ValueError                                Traceback (most recent call last)
~/anaconda3/envs/python3/lib/python3.6/site-packages/partd/numpy.py in deserialize(bytes, dtype, copy)
    112             blocks = [msgpack.unpackb(f, **unpack_kwargs)
--> 113                       for f in framesplit(bytes)]
    114         except Exception:

~/anaconda3/envs/python3/lib/python3.6/site-packages/partd/numpy.py in <listcomp>(.0)
    112             blocks = [msgpack.unpackb(f, **unpack_kwargs)
--> 113                       for f in framesplit(bytes)]
    114         except Exception:

msgpack/_unpacker.pyx in msgpack._cmsgpack.unpackb()

ValueError: 604400 exceeds max_array_len(131072)

During handling of the above exception, another exception occurred:

UnpicklingError                           Traceback (most recent call last)
.....
.....
UnpicklingError: invalid load key, '\xdd'.

The pandas version is 1.0.3.
The dask version is 2.13.0.

[gforsyth]

Hey @RokoMijic -- can you open a new issue for this? Thanks!

[RokoMijic]

Hey @RokoMijic -- can you open a new issue for this? Thanks!

Do you need me to make an example that reproduces it?

[jrbourbeau]

A minimal reproducible example (https://blog.dask.org/2018/02/28/minimal-bug-reports) would be great, thanks @RokoMijic!

[RokoMijic]

OK, I'll see whether I can do it.

I eventually resolved the problem by writing my own outer merge function and doing it in Pandas, as Dask was unstable and crashed every time.

https://stackoverflow.com/questions/61026417/how-do-you-efficiently-outer-merge-large-pandas-dataframes-whilst-preserving-dat/