Create new partd for every bag groupby (#1867)

Fixes #1851
dask · Dec 26, 2016 · 76fffb2 · 76fffb2
1 parent b2b3444
commit 76fffb2
Show file tree

Hide file tree

Showing 2 changed files with 8 additions and 2 deletions.
diff --git a/dask/bag/core.py b/dask/bag/core.py
@@ -1590,9 +1590,9 @@ def groupby_disk(b, grouper, npartitions=None, blocksize=2**20):
     import partd
     p = ('partd-' + token,)
     try:
-        dsk1 = {p: (partd.Python, (partd.Snappy, partd.File()))}
+        dsk1 = {p: (partd.Python, (partd.Snappy, (partd.File,)))}
     except AttributeError:
-        dsk1 = {p: (partd.Python, partd.File())}
+        dsk1 = {p: (partd.Python, (partd.File,))}
 
     # Partition data on disk
     name = 'groupby-part-{0}-{1}'.format(funcname(grouper), token)

diff --git a/dask/bag/tests/test_bag.py b/dask/bag/tests/test_bag.py
@@ -1100,3 +1100,9 @@ def func(part):
     res = b.reduction(func, sum)
 
     assert res.compute(get=dask.get) == sum(range(10))
+
+
+def test_repeated_groupby():
+    b = db.range(10, npartitions=4)
+    c = b.groupby(lambda x: x % 3)
+    assert valmap(len, dict(c)) == valmap(len, dict(c))