Repository: dask/dask
Pull request: dask.array.bincount() - make 'minlength' keyword argument optional
This PR would make the `minlength` keyword argument to `dask.array.bincount()` optional, instead of required.

It implements @mrocklin 's [suggested approach](https://github.com/dask/dask/pull/4683#pullrequestreview-224744657):

> As an alternative, what if we just called `np.bincount` on all of the chunks without specifying a minlength value and only afterwards, once we had all of them together, would we make a new array of the correct length and then add all of them into that array? This might require a bit more book keeping as we add things together (it's no longer a simple `np.sum` I think) but would probably improve our ability to keep this running smoothly in low memory.

If we can think of a nicer way to sum a bunch of jagged arrays together, I'm all ears. Right now I'm zero-padding things to the same length, but maybe there's a neater way?

- [x] Tests added / passed
- [x] Passes `flake8 dask`

Closes https://github.com/dask/dask/issues/4556


Problem statement:
`bincount` assumes `minlength` is the length
Currently Dask Array's `bincount`  assumes `minlength` is the length of the underlying array. Unfortunately, as noted in NumPy's docs, "if minlength is specified, there will be at least this number of bins in the output array (though it will be longer if necessary, depending on the contents of x)." IOW `minlength` is not the length of the array only the lower bound. This issue can be demonstrated by the following example.

<details>

```python
In [1]: import numpy as np                                                      

In [2]: import dask.array as da                                                 

In [3]: a = np.asarray([0, 1, 2, 3])                                            

In [4]: d = da.from_array(a, chunks=2)                                          

In [5]: r = da.bincount(d, minlength=1)                                         

In [6]: r                                                                       
Out[6]: dask.array<bincount-sum, shape=(1,), dtype=int64, chunksize=(1,)>

In [7]: r.compute()                                                             
---------------------------------------------------------------------------
ValueError                                Traceback (most recent call last)
<ipython-input-7-8c9e270be48d> in <module>
----> 1 r.compute()

/zopt/conda3/envs/nanshenv3/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 

/zopt/conda3/envs/nanshenv3/lib/python3.6/site-packages/dask/base.py in compute(*args, **kwargs)
    395     keys = [x.__dask_keys__() for x in collections]
    396     postcomputes = [x.__dask_postcompute__() for x in collections]
--> 397     results = schedule(dsk, keys, **kwargs)
    398     return repack([f(r, *a) for r, (f, a) in zip(results, postcomputes)])
    399 

/zopt/conda3/envs/nanshenv3/lib/python3.6/site-packages/dask/threaded.py in get(dsk, result, cache, num_workers, **kwargs)
     74     results = get_async(pool.apply_async, len(pool._pool), dsk, result,
     75                         cache=cache, get_id=_thread_get_id,
---> 76                         pack_exception=pack_exception, **kwargs)
     77 
     78     # Cleanup pools associated to dead threads

/zopt/conda3/envs/nanshenv3/lib/python3.6/site-packages/dask/local.py in get_async(apply_async, num_workers, dsk, result, cache, get_id, rerun_exceptions_locally, pack_exception, raise_exception, callbacks, dumps, loads, **kwargs)
    499                         _execute_task(task, data)  # Re-execute locally
    500                     else:
--> 501                         raise_exception(exc, tb)
    502                 res, worker_id = loads(res_info)
    503                 state['cache'][key] = res

/zopt/conda3/envs/nanshenv3/lib/python3.6/site-packages/dask/compatibility.py in reraise(exc, tb)
    110         if exc.__traceback__ is not tb:
    111             raise exc.with_traceback(tb)
--> 112         raise exc
    113 
    114 else:

/zopt/conda3/envs/nanshenv3/lib/python3.6/site-packages/dask/local.py in execute_task(key, task_info, dumps, loads, get_id, pack_exception)
    270     try:
    271         task, data = loads(task_info)
--> 272         result = _execute_task(task, data)
    273         id = get_id()
    274         result = dumps((result, id))

/zopt/conda3/envs/nanshenv3/lib/python3.6/site-packages/dask/local.py in _execute_task(arg, cache, dsk)
    251         func, args = arg[0], arg[1:]
    252         args2 = [_execute_task(a, cache) for a in args]
--> 253         return func(*args2)
    254     elif not ishashable(arg):
    255         return arg

/zopt/conda3/envs/nanshenv3/lib/python3.6/site-packages/numpy/core/fromnumeric.py in sum(a, axis, dtype, out, keepdims, initial)
   1928 
   1929     return _wrapreduction(a, np.add, 'sum', axis, dtype, out, keepdims=keepdims,
-> 1930                           initial=initial)
   1931 
   1932 

/zopt/conda3/envs/nanshenv3/lib/python3.6/site-packages/numpy/core/fromnumeric.py in _wrapreduction(obj, ufunc, method, axis, dtype, out, **kwargs)
     81                 return reduction(axis=axis, out=out, **passkwargs)
     82 
---> 83     return ufunc.reduce(obj, axis, dtype, out, **passkwargs)
     84 
     85 

ValueError: operands could not be broadcast together with shapes (2,) (4,) 
```

</details>

<br>

Further `bincount` does not currently work when `minlength` is not specified (see below). This is likely built on the assumption that the array length is known and should be `minlength`. In reality, the array size is unknown and dependent on the computation.

<details>

```python
In [1]: import numpy as np                                                      

In [2]: import dask.array as da                                                 

In [3]: a = np.asarray([0, 1, 2, 3])                                            

In [4]: d = da.from_array(a, chunks=2)                                          

In [5]: da.bincount(d)                                                          
---------------------------------------------------------------------------
TypeError                                 Traceback (most recent call last)
<ipython-input-5-4257836e61f5> in <module>
----> 1 da.bincount(d)

/zopt/conda3/envs/nanshenv3/lib/python3.6/site-packages/dask/array/routines.py in bincount(x, weights, minlength)
    528 def bincount(x, weights=None, minlength=None):
    529     if minlength is None:
--> 530         raise TypeError("Must specify minlength argument in da.bincount")
    531     assert x.ndim == 1
    532     if weights is not None:

TypeError: Must specify minlength argument in da.bincount
```

</details>

<br>

A reasonable solution going forward would be to set the chunk size to `nan` as they are unknown. There may be a bit of work under the hood to handle reductions where the lengths of the two chunks don't match (e.g. padding with zeros) as evidenced by the first failure mode. There may be a bit more work to handle `nan` chunk sizes in reductions generally (though this is less clear).

Hints from issue discussion:
cc @GenevieveBuckley
I've played around with da.bincount() a bit, and feel like I sometimes got inconsistent output regarding minlength. Maybe this was due to switching between different versions of dask (or even numpy?) at the time? 

I don't really remember (and don't have time to dig into it either), but wanted to mention that here in case it helps someone else.
cc-ing @rainwoodman (as it looks like you have been thinking about `bincount` recently and it would be good to hear your thoughts here generally :)
I usually assumed min_length is the length, and never relied on the numpy behavior to increase the length.

If we must preserve numpy compatibility, then I feel a two stage approach, first computing the true length, then properly setting the chunk size may be better than using nan chunksizes. Although it does mean streaming through the input twice, returning a properly chunked result is more useful than a limitedly supported nan chunked array.

I feel the numpy logic here is a bit awkward -- it never returns a predictably sized array; while expecting a predictably sized result is a frequent use case. 

We can therefore continue to give up on strict numpy compatibility here. Just need to make the error message more informative if we encounter any item larger than minlength.




Stepping back a bit, is this the operation that people actually want?  Would something like approximate quantiles work as well here?  We can also extend beyond the NumPy API if appropriate.
Could you clarify `this` you mean?
>  is this the operation that people actually want?

bincount I guess.  I think that there have been a few issues about histogram-like operations recently.  My comment here was probably slightly inaccurate in that I'm thinking about all of them, rather than just this one issue.  I could also be confused though.
In that case, bincount can be used to put particles into bins on a 'irregular' mesh -- the exactness of bincount is necessary. 
Our use case is computing the area of objects in a label image (created using [connected components]( https://homepages.inf.ed.ac.uk/rbf/HIPR2/label.htm )). So `bincount` would be a nice fit as opposed to other operations as the data is already discretized in units that we want to preserve. Whereas quantiles and histograms make more sense (at least to me) when this discretization has not already occurred or does not match with the groups we want in the end. There certainly are use cases for these strategies in image processing as well, but they are different from this one.
To reproduce numpy-like behaviour, we'd need `minlength = np.max(input_dask_array) + 1`. This bit has to happen first in the task graph, since we need an integer result for the `minlength` keyword argument in bincount.

This seems relevant. The dask [specification docs](https://docs.dask.org/en/latest/spec.html) say: 
> To encode keyword arguments we recommend the use of ``functools.partial`` or
``toolz.curry``.

Is this the type of use case that suggestion is made for?

Commit message: Bump pandas to 0.20.2 for python 3.5 travis build

Patch to review:
diff --git a/.travis.yml b/.travis.yml
index 7de0412b195..3baa0a1e109 100644
--- a/.travis.yml
+++ b/.travis.yml
@@ -27,8 +27,8 @@ jobs:
 
     - env:
       - PYTHON=3.5
-      - NUMPY=1.11.1
-      - PANDAS=0.19.2
+      - NUMPY=1.13.0
+      - PANDAS=0.20.2
       - *test_and_lint
       - *no_coverage
       - *no_optimize
diff --git a/dask/array/routines.py b/dask/array/routines.py
index 21925d84ebe..1387d2d0030 100644
--- a/dask/array/routines.py
+++ b/dask/array/routines.py
@@ -523,34 +523,44 @@ def gradient(f, *varargs, **kwargs):
     return results
 
 
+def _bincount_sum(bincounts, dtype=int):
+    n = max(map(len, bincounts))
+    out = np.zeros(n, dtype=dtype)
+    for b in bincounts:
+        out[:len(b)] += b[:len(b)]
+    return out
+
+
 @wraps(np.bincount)
-def bincount(x, weights=None, minlength=None):
-    if minlength is None:
-        raise TypeError("Must specify minlength argument in da.bincount")
+def bincount(x, weights=None, minlength=0):
     assert x.ndim == 1
     if weights is not None:
         assert weights.chunks == x.chunks
 
     # Call np.bincount on each block, possibly with weights
+    dsk = {}
     token = tokenize(x, weights, minlength)
-    name = 'bincount-' + token
-    if weights is not None:
-        dsk = {(name, i): (np.bincount, (x.name, i), (weights.name, i), minlength)
-               for i, _ in enumerate(x.__dask_keys__())}
-        dtype = np.bincount([1], weights=[1]).dtype
-    else:
-        dsk = {(name, i): (np.bincount, (x.name, i), None, minlength)
-               for i, _ in enumerate(x.__dask_keys__())}
-        dtype = np.bincount([]).dtype
+    for i, _ in enumerate(x.__dask_keys__()):
+        if weights is not None:
+            dsk[('bincount-' + token, i)] = (np.bincount, (x.name, i),
+                                             (weights.name, i), minlength)
+            dtype = np.bincount([1], weights=[1]).dtype
+        else:
+            dsk[('bincount-' + token, i)] = (np.bincount, (x.name, i),
+                                             None, minlength)
+            dtype = np.bincount([]).dtype
 
-    # Sum up all of the intermediate bincounts per block
+    bincount_list = [i for i in list(dsk) if i[0] == 'bincount-' + token]
     name = 'bincount-sum-' + token
-    dsk[(name, 0)] = (np.sum, list(dsk), 0)
-
-    chunks = ((minlength,),)
+    dsk[(name, 0)] = (_bincount_sum, bincount_list, dtype)
 
     graph = HighLevelGraph.from_collections(name, dsk, dependencies=[x] if weights is None else [x, weights])
 
+    if minlength == 0:
+        chunks = ((np.nan,),)
+    else:
+        chunks = ((minlength,),)
+
     return Array(graph, name, chunks, dtype)
 
 
diff --git a/dask/array/tests/test_routines.py b/dask/array/tests/test_routines.py
index 290f958ce70..bcedadec075 100644
--- a/dask/array/tests/test_routines.py
+++ b/dask/array/tests/test_routines.py
@@ -506,15 +506,13 @@ def test_bincount_with_weights():
     assert same_keys(da.bincount(d, weights=dweights, minlength=6), e)
 
 
-def test_bincount_raises_informative_error_on_missing_minlength_kwarg():
-    x = np.array([2, 1, 5, 2, 1])
+def test_bincount_unspecified_minlength():
+    x = np.array([1, 1, 3, 7, 0])
     d = da.from_array(x, chunks=2)
-    try:
-        da.bincount(d)
-    except Exception as e:
-        assert 'minlength' in str(e)
-    else:
-        assert False
+    e = da.bincount(d)
+    assert_eq(e, np.bincount(x))
+    assert same_keys(da.bincount(d), e)
+    assert len(e.compute()) == 8  # shape is (nan,) so must compute for len()
 
 
 def test_digitize():


Code review comments:
diff_hunk: @@ -523,34 +523,44 @@ def gradient(f, *varargs, **kwargs):
     return results
 
 
+def _bincount_sum(bincounts, dtype=int):
+    n = max(map(len, bincounts))
+    out = np.zeros(n, dtype=dtype)
+    for b in bincounts:
+        out[:len(b)] += b[:len(b)]
+    return out
+
+
 @wraps(np.bincount)
-def bincount(x, weights=None, minlength=None):
-    if minlength is None:
-        raise TypeError("Must specify minlength argument in da.bincount")
+def bincount(x, weights=None, minlength=0):
     assert x.ndim == 1
     if weights is not None:
         assert weights.chunks == x.chunks
 
     # Call np.bincount on each block, possibly with weights
+    dsk = {}
     token = tokenize(x, weights, minlength)
-    name = 'bincount-' + token
-    if weights is not None:
-        dsk = {(name, i): (np.bincount, (x.name, i), (weights.name, i), minlength)
-               for i, _ in enumerate(x.__dask_keys__())}
-        dtype = np.bincount([1], weights=[1]).dtype
-    else:
-        dsk = {(name, i): (np.bincount, (x.name, i), None, minlength)
-               for i, _ in enumerate(x.__dask_keys__())}
-        dtype = np.bincount([]).dtype
+    for i, _ in enumerate(x.__dask_keys__()):
+        if weights is not None:
+            dsk[('bincount-' + token, i)] = (np.bincount, (x.name, i),
+                                             (weights.name, i), minlength)
+            dtype = np.bincount([1], weights=[1]).dtype
+        else:
+            dsk[('bincount-' + token, i)] = (np.bincount, (x.name, i),
+                                             None, minlength)
+            dtype = np.bincount([]).dtype
 
-    # Sum up all of the intermediate bincounts per block
+    bincount_list = [i for i in list(dsk) if i[0] == 'bincount-' + token]
     name = 'bincount-sum-' + token
-    dsk[(name, 0)] = (np.sum, list(dsk), 0)
-
-    chunks = ((minlength,),)
+    dsk[(name, 0)] = (_bincount_sum, bincount_list, dtype)
path: dask/array/routines.py
text: @user1:
```suggestion
    dsk[(name, 0)] = (_bincount_sum, list(dsk), dtype)
```

@user1:
I think that we can now remove the `bincount_list` variable.

diff_hunk: @@ -523,34 +523,44 @@ def gradient(f, *varargs, **kwargs):
     return results
 
 
+def _bincount_sum(bincounts, dtype=int):
+    n = max(map(len, bincounts))
+    out = np.zeros(n, dtype=dtype)
+    for b in bincounts:
+        out[:len(b)] += b[:len(b)]
+    return out
+
+
 @wraps(np.bincount)
-def bincount(x, weights=None, minlength=None):
-    if minlength is None:
-        raise TypeError("Must specify minlength argument in da.bincount")
+def bincount(x, weights=None, minlength=0):
     assert x.ndim == 1
     if weights is not None:
         assert weights.chunks == x.chunks
 
     # Call np.bincount on each block, possibly with weights
+    dsk = {}
     token = tokenize(x, weights, minlength)
-    name = 'bincount-' + token
-    if weights is not None:
-        dsk = {(name, i): (np.bincount, (x.name, i), (weights.name, i), minlength)
-               for i, _ in enumerate(x.__dask_keys__())}
-        dtype = np.bincount([1], weights=[1]).dtype
-    else:
-        dsk = {(name, i): (np.bincount, (x.name, i), None, minlength)
-               for i, _ in enumerate(x.__dask_keys__())}
-        dtype = np.bincount([]).dtype
+    for i, _ in enumerate(x.__dask_keys__()):
+        if weights is not None:
+            dsk[('bincount-' + token, i)] = (np.bincount, (x.name, i),
+                                             (weights.name, i), minlength)
+            dtype = np.bincount([1], weights=[1]).dtype
+        else:
+            dsk[('bincount-' + token, i)] = (np.bincount, (x.name, i),
+                                             None, minlength)
+            dtype = np.bincount([]).dtype
 
-    # Sum up all of the intermediate bincounts per block
+    bincount_list = [i for i in list(dsk) if i[0] == 'bincount-' + token]
     name = 'bincount-sum-' + token
path: dask/array/routines.py
text: @user1:
Maybe we can move the definition of `name` up by `tokenize` again (maybe even removing the `token` intermediate variable)

@author:
Sure. I don't really understand what what token is doing here, tbh

@author:
`token` variable has been removed.

@user1:
Ah, sorry, the tokenize process is still improtant.  Every task is uniquely named in Dask, similar to git commits, to help with deduplication.  I've reverted this and changed things a little bit in https://github.com/dask/dask/pull/4684/commits/41c6b00f96a3c0c11a6d374ce9500ee24fa912c3 and added tests to show why it's used. 

@author:
Aaah, ok. I had assumed that was what it was for, but then got confused about what you were trying to say with this comment. Thanks for clarifying!

diff_hunk: @@ -523,34 +523,44 @@ def gradient(f, *varargs, **kwargs):
     return results
 
 
+def _bincount_sum(bincounts, dtype=int):
+    n = max(map(len, bincounts))
+    out = np.zeros(n, dtype=dtype)
+    for b in bincounts:
+        out[:len(b)] += b[:len(b)]
+    return out
+
+
 @wraps(np.bincount)
-def bincount(x, weights=None, minlength=None):
-    if minlength is None:
-        raise TypeError("Must specify minlength argument in da.bincount")
+def bincount(x, weights=None, minlength=0):
     assert x.ndim == 1
     if weights is not None:
         assert weights.chunks == x.chunks
 
     # Call np.bincount on each block, possibly with weights
+    dsk = {}
     token = tokenize(x, weights, minlength)
-    name = 'bincount-' + token
-    if weights is not None:
-        dsk = {(name, i): (np.bincount, (x.name, i), (weights.name, i), minlength)
-               for i, _ in enumerate(x.__dask_keys__())}
-        dtype = np.bincount([1], weights=[1]).dtype
-    else:
-        dsk = {(name, i): (np.bincount, (x.name, i), None, minlength)
-               for i, _ in enumerate(x.__dask_keys__())}
-        dtype = np.bincount([]).dtype
+    for i, _ in enumerate(x.__dask_keys__()):
+        if weights is not None:
+            dsk[('bincount-' + token, i)] = (np.bincount, (x.name, i),
+                                             (weights.name, i), minlength)
+            dtype = np.bincount([1], weights=[1]).dtype
+        else:
+            dsk[('bincount-' + token, i)] = (np.bincount, (x.name, i),
+                                             None, minlength)
+            dtype = np.bincount([]).dtype
 
-    # Sum up all of the intermediate bincounts per block
+    bincount_list = [i for i in list(dsk) if i[0] == 'bincount-' + token]
path: dask/array/routines.py
text: @user1:
```suggestion
```