Support dask.bag reductions on no partitions

[mrocklin]

No description provided.

[jcrist]

I may be wrong, but I think this patch is equivalent, but doesn't special case certain numbers of partitions:

--- a/dask/bag/core.py
+++ b/dask/bag/core.py
@@ -731,11 +731,10 @@ class Bag(Base):
         b = a
         fmt = '%s-aggregate-%s' % (name or funcname(aggregate), token)
         depth = 0
-        while k > 1:
+        while k > split_every:
             c = fmt + str(depth)
-            is_last = k <= split_every
             dsk2 = dict(((c, i), (empty_safe_aggregate, aggregate,
-                                  [(b, j) for j in inds], is_last))
+                                  [(b, j) for j in inds], False))
                         for i, inds in enumerate(partition_all(split_every,
                                                                range(k))))
             dsk.update(dsk2)
@@ -743,11 +742,14 @@ class Bag(Base):
             b = c
             depth += 1

+        dsk[(fmt, 0)] = (empty_safe_aggregate, aggregate,
+                         [(b, j) for j in range(k)], True)
+
         if out_type is Item:
-            dsk[b] = dsk.pop((b, 0))
-            return Item(merge(self.dask, dsk), b)
+            dsk[fmt] = dsk.pop((fmt, 0))
+            return Item(merge(self.dask, dsk), fmt)
         else:
-            return Bag(merge(self.dask, dsk), b, 1)
+            return Bag(merge(self.dask, dsk), fmt, 1)

     def sum(self, split_every=None):
         """ Sum all elements """

[mrocklin]

Merging this soon if there are no further comments

[jcrist]

Looks good to me.