Repository: dask/dask

Issue:
Bag groupby error using tasks based shuffle
Looks like the issue is when the bag in question has only 1 partition.

```
>>> dask.bag.from_sequence('abc', npartitions = 1).groupby(lambda x: x, shuffle = 'tasks').compute()
Traceback (most recent call last):
  File "<stdin>", line 1, in <module>
  File "/usr/local/lib/python3.6/dist-packages/dask/bag/core.py", line 1240, in groupby
    return groupby_tasks(self, grouper, max_branch=max_branch)
  File "/usr/local/lib/python3.6/dist-packages/dask/bag/core.py", line 2012, in groupby_tasks
    for i, j in enumerate(join))
UnboundLocalError: local variable 'join' referenced before assignment
```

Dask version 0.18.1

Discussion:

Patch:
diff --git a/dask/bag/core.py b/dask/bag/core.py
index bc79ee350..d8bf1c6eb 100644
--- a/dask/bag/core.py
+++ b/dask/bag/core.py
@@ -1970,7 +1970,7 @@ def groupby_tasks(b, grouper, hash=hash, max_branch=32):
     max_branch = max_branch or 32
     n = b.npartitions
 
-    stages = int(math.ceil(math.log(n) / math.log(max_branch)))
+    stages = int(math.ceil(math.log(n) / math.log(max_branch))) or 1
     if stages > 1:
         k = int(math.ceil(n ** (1 / stages)))
     else:


Test patch:
diff --git a/dask/bag/tests/test_bag.py b/dask/bag/tests/test_bag.py
index fa93e5956..1c214314d 100644
--- a/dask/bag/tests/test_bag.py
+++ b/dask/bag/tests/test_bag.py
@@ -1138,7 +1138,8 @@ def test_groupby_tasks_names():
 
 
 @pytest.mark.parametrize('size,npartitions,groups', [(1000, 20, 100),
-                                                     (12345, 234, 1042)])
+                                                     (12345, 234, 1042),
+                                                     (100, 1, 50)])
 def test_groupby_tasks_2(size, npartitions, groups):
     func = lambda x: x % groups
     b = db.range(size, npartitions=npartitions).groupby(func, shuffle='tasks')