Repository: dask/dask

Issue:
dask.bag does not use hashed keys
We should use `tokenize` rather than the current `tokens` within dask.bag

cc @jcrist 

Discussion:

Patch:
diff --git a/dask/bag/core.py b/dask/bag/core.py
index fcc3119fb..2c4c9115b 100644
--- a/dask/bag/core.py
+++ b/dask/bag/core.py
@@ -5,11 +5,11 @@ import itertools
 import math
 import bz2
 import os
+import uuid
 from fnmatch import fnmatchcase
 from glob import glob
 from collections import Iterable, Iterator, defaultdict
 from functools import wraps, partial
-from itertools import repeat
 
 from ..utils import ignoring
 
@@ -23,16 +23,12 @@ with ignoring(ImportError):
 
 from ..base import Base, normalize_token, tokenize
 from ..compatibility import (apply, BytesIO, unicode, urlopen, urlparse,
-        StringIO, GzipFile, BZ2File)
+        GzipFile)
 from ..core import list2, quote, istask, get_dependencies, reverse_dict
 from ..multiprocessing import get as mpget
 from ..optimize import fuse, cull, inline
 from ..utils import (file_size, infer_compression, open, system_encoding,
-                     takes_multiple_arguments, textblock)
-
-names = ('bag-%d' % i for i in itertools.count(1))
-tokens = ('-%d' % i for i in itertools.count(1))
-load_names = ('load-%d' % i for i in itertools.count(1))
+                     takes_multiple_arguments, textblock, funcname)
 
 no_default = '__no__default__'
 
@@ -164,7 +160,7 @@ def to_textfiles(b, path, name_function=str, compression='infer',
             compression = infer_compression(path)
         return compression
 
-    name = next(names)
+    name = 'to-textfiles-' + uuid.uuid4().hex
     dsk = dict(((name, i), (write, (b.name, i), path, get_compression(path),
                             encoding))
                for i, path in enumerate(paths))
@@ -199,7 +195,7 @@ class Item(Base):
         return [self.key]
 
     def apply(self, func):
-        name = next(names)
+        name = 'apply-{0}-{1}'.format(funcname(func), tokenize(self, func))
         dsk = {name: (func, self.key)}
         return Item(merge(self.dask, dsk), name)
 
@@ -254,7 +250,7 @@ class Bag(Base):
         >>> list(b.map(lambda x: x * 10))  # doctest: +SKIP
         [0, 10, 20, 30, 40]
         """
-        name = next(names)
+        name = 'map-{0}-{1}'.format(funcname(func), tokenize(self, func))
         if takes_multiple_arguments(func):
             func = partial(apply, func)
         dsk = dict(((name, i), (reify, (map, func, (self.name, i))))
@@ -276,7 +272,8 @@ class Bag(Base):
         >>> list(b.filter(iseven))  # doctest: +SKIP
         [0, 2, 4]
         """
-        name = next(names)
+        name = 'filter-{0}-{1}'.format(funcname(predicate),
+                                       tokenize(self, predicate))
         dsk = dict(((name, i), (reify, (filter, predicate, (self.name, i))))
                    for i in range(self.npartitions))
         return type(self)(merge(self.dask, dsk), name, self.npartitions)
@@ -292,7 +289,8 @@ class Bag(Base):
         >>> list(b.remove(iseven))  # doctest: +SKIP
         [1, 3]
         """
-        name = next(names)
+        name = 'remove-{0}-{1}'.format(funcname(predicate),
+                                       tokenize(self, predicate))
         dsk = dict(((name, i), (reify, (remove, predicate, (self.name, i))))
                    for i in range(self.npartitions))
         return type(self)(merge(self.dask, dsk), name, self.npartitions)
@@ -305,7 +303,8 @@ class Bag(Base):
 
         >>> b.map_partitions(myfunc)  # doctest: +SKIP
         """
-        name = next(names)
+        name = 'map-partitions-{0}-{1}'.format(funcname(func),
+                                               tokenize(self, func))
         dsk = dict(((name, i), (func, (self.name, i)))
                    for i in range(self.npartitions))
         return type(self)(merge(self.dask, dsk), name, self.npartitions)
@@ -320,7 +319,7 @@ class Bag(Base):
         >>> list(b.pluck('credits').pluck(0))  # doctest: +SKIP
         [1, 10]
         """
-        name = next(names)
+        name = 'pluck-' + tokenize(self, key, default)
         key = quote(key)
         if default == no_default:
             dsk = dict(((name, i), (list, (pluck, key, (self.name, i))))
@@ -391,8 +390,10 @@ class Bag(Base):
 
         Bag.foldby
         """
-        a = next(names)
-        b = next(names)
+        token = tokenize(self, binop, combine, initial)
+        combine = combine or binop
+        a = 'foldbinop-{0}-{1}'.format(funcname(binop), token)
+        b = 'foldcombine-{0}-{1}'.format(funcname(combine), token)
         initial = quote(initial)
         if initial is not no_default:
             dsk = dict(((a, i), (reduce, binop, (self.name, i), initial))
@@ -400,7 +401,7 @@ class Bag(Base):
         else:
             dsk = dict(((a, i), (reduce, binop, (self.name, i)))
                        for i in range(self.npartitions))
-        dsk2 = {b: (reduce, combine or binop, list(dsk.keys()))}
+        dsk2 = {b: (reduce, combine, list(dsk.keys()))}
         return Item(merge(self.dask, dsk, dsk2), b)
 
     def frequencies(self, split_every=None):
@@ -426,8 +427,9 @@ class Bag(Base):
         >>> list(b.topk(2, lambda x: -x))  # doctest: +SKIP
         [3, 4]
         """
-        a = next(names)
-        b = next(names)
+        token = tokenize(self, k, key)
+        a = 'topk-a-' + token
+        b = 'topk-b-' + token
         if key:
             if callable(key) and takes_multiple_arguments(key):
                 key = partial(apply, key)
@@ -448,9 +450,10 @@ class Bag(Base):
         >>> sorted(b.distinct())
         ['Alice', 'Bob']
         """
-        a = next(names)
+        token = tokenize(self)
+        a = 'distinct-a-' + token
+        b = 'distinct-b-' + token
         dsk = dict(((a, i), (set, key)) for i, key in enumerate(self._keys()))
-        b = next(names)
         dsk2 = {(b, 0): (apply, set.union, quote(list(dsk.keys())))}
 
         return type(self)(merge(self.dask, dsk, dsk2), b, 1)
@@ -482,19 +485,22 @@ class Bag(Base):
         """
         if split_every is None:
             split_every = 8
-        a = next(names)
-        b = next(names)
+        token = tokenize(self, perpartition, aggregate, split_every)
+        a = 'reduction-part-{0}-{1}'.format(funcname(perpartition), token)
         dsk = dict(((a, i), (perpartition, (self.name, i)))
                    for i in range(self.npartitions))
         k = self.npartitions
         b = a
+        fmt = 'reduction-agg-{0}-'.format(funcname(aggregate)) + '-{0}-' + token
+        depth = 0
         while k > 1:
-            c = next(names)
+            c = fmt.format(depth)
             dsk2 = dict(((c, i), (aggregate, [(b, j) for j in inds]))
                  for i, inds in enumerate(partition_all(split_every, range(k))))
             dsk.update(dsk2)
             k = len(dsk2)
             b = c
+            depth += 1
 
         if out_type is Item:
             dsk[c] = dsk.pop((c, 0))
@@ -502,7 +508,6 @@ class Bag(Base):
         else:
             return Bag(merge(self.dask, dsk), c, 1)
 
-
     @wraps(sum)
     def sum(self, split_every=None):
         return self.reduction(sum, sum, split_every=split_every)
@@ -578,7 +583,7 @@ class Bag(Base):
         assert not isinstance(other, Bag)
         if on_other is None:
             on_other = on_self
-        name = next(names)
+        name = 'join-' + tokenize(self, other, on_self, on_other)
         dsk = dict(((name, i), (list, (join, on_other, other,
                                        on_self, (self.name, i))))
                    for i in range(self.npartitions))
@@ -587,7 +592,7 @@ class Bag(Base):
     def product(self, other):
         """ Cartesian product between two bags """
         assert isinstance(other, Bag)
-        name = next(names)
+        name = 'product-' + tokenize(self, other)
         n, m = self.npartitions, other.npartitions
         dsk = dict(((name, i*m + j),
                    (list, (itertools.product, (self.name, i),
@@ -667,8 +672,9 @@ class Bag(Base):
         toolz.reduceby
         pyspark.combineByKey
         """
-        a = next(names)
-        b = next(names)
+        token = tokenize(self, key, binop, initial, combine, combine_initial)
+        a = 'foldby-a-' + token
+        b = 'foldby-b-' + token
         if combine is None:
             combine = binop
         if initial is not no_default:
@@ -705,7 +711,7 @@ class Bag(Base):
         >>> b.take(3)  # doctest: +SKIP
         (0, 1, 2)
         """
-        name = next(names)
+        name = 'take-' + tokenize(self, k)
         dsk = {(name, 0): (list, (take, k, (self.name, 0)))}
         b = Bag(merge(self.dask, dsk), name, 1)
         if compute:
@@ -726,7 +732,7 @@ class Bag(Base):
         >>> list(b.concat())
         [1, 2, 3]
         """
-        name = next(names)
+        name = 'concat-' + tokenize(self)
         dsk = dict(((name, i), (list, (toolz.concat, (self.name, i))))
                    for i in range(self.npartitions))
         return type(self)(merge(self.dask, dsk), name, self.npartitions)
@@ -751,22 +757,23 @@ class Bag(Base):
         """
         if npartitions is None:
             npartitions = self.npartitions
+        token = tokenize(self, grouper, npartitions, blocksize)
 
         import partd
-        p = ('partd' + next(tokens),)
+        p = ('partd-' + token,)
         try:
             dsk1 = {p: (partd.Python, (partd.Snappy, partd.File()))}
         except AttributeError:
             dsk1 = {p: (partd.Python, partd.File())}
 
         # Partition data on disk
-        name = next(names)
+        name = 'groupby-part-{0}-{1}'.format(funcname(grouper), token)
         dsk2 = dict(((name, i), (partition, grouper, (self.name, i),
                                  npartitions, p, blocksize))
                     for i in range(self.npartitions))
 
         # Barrier
-        barrier_token = 'barrier' + next(tokens)
+        barrier_token = 'groupby-barrier-' + token
 
         def barrier(args):
             return 0
@@ -774,7 +781,7 @@ class Bag(Base):
         dsk3 = {barrier_token: (barrier, list(dsk2))}
 
         # Collect groups
-        name = next(names)
+        name = 'groupby-collect-' + token
         dsk4 = dict(((name, i),
                      (collect, grouper, i, p, barrier_token))
                     for i in range(npartitions))
@@ -816,7 +823,7 @@ class Bag(Base):
                 columns = sorted(head)
             elif isinstance(head, (tuple, list)):
                 columns = list(range(len(head)))
-        name = next(names)
+        name = 'to_dataframe-' + tokenize(self, columns)
         DataFrame = partial(pd.DataFrame, columns=columns)
         dsk = dict(((name, i), (DataFrame, (list2, (self.name, i))))
                    for i in range(self.npartitions))
@@ -885,7 +892,7 @@ def from_filenames(filenames, chunkbytes=None, compression='infer',
 
     full_filenames = [os.path.abspath(f) for f in filenames]
 
-    name = 'from-filename' + next(tokens)
+    name = 'from-filenames-' + uuid.uuid4().hex
 
     # Make sure `linesep` is not a byte string because `io.TextIOWrapper` in
     # python versions other than 2.7 dislike byte strings for the `newline`
@@ -992,7 +999,7 @@ def from_s3(bucket_name, paths='*', aws_access_key=None, aws_secret_key=None,
 
     get_key = partial(_get_key, bucket_name, conn_args)
 
-    name = next(load_names)
+    name = 'from_s3-' + uuid.uuid4().hex
     dsk = dict(((name, i), (list, (get_key, k))) for i, k in enumerate(paths))
     return Bag(dsk, name, len(paths))
 
@@ -1035,7 +1042,7 @@ def from_hdfs(path, hdfs=None, host='localhost', port='50070', user_name=None):
     if not filenames:
         raise ValueError("No files found for path %s" % path)
 
-    name = next(names)
+    name = 'from_hdfs-' + uuid.uuid4().hex
     dsk = dict()
     for i, fn in enumerate(filenames):
         ext = fn.split('.')[-1]
@@ -1106,7 +1113,7 @@ def from_sequence(seq, partition_size=None, npartitions=None):
             partition_size = int(len(seq) / 100)
 
     parts = list(partition_all(partition_size, seq))
-    name = next(load_names)
+    name = 'from_sequence-' + tokenize(seq, partition_size)
     d = dict(((name, i), part) for i, part in enumerate(parts))
     return Bag(d, name, len(d))
 
@@ -1131,7 +1138,8 @@ def from_castra(x, columns=None, index=False):
     if columns is None:
         columns = x.columns
 
-    name = 'from-castra-' + next(tokens)
+    name = 'from-castra-' + tokenize(os.path.getmtime(x.path), x.path,
+                                     columns, index)
     dsk = dict(((name, i), (load_castra_partition, x, part, columns, index))
                 for i, part in enumerate(x.partitions))
     return Bag(dsk, name, len(x.partitions))
@@ -1181,7 +1189,7 @@ def from_url(urls):
     """
     if isinstance(urls, str):
         urls = [urls]
-    name = next(load_names)
+    name = 'from_url-' + uuid.uuid4().hex
     dsk = {}
     for i, u in enumerate(urls):
         dsk[(name, i)] = (list, (urlopen, u))
@@ -1208,7 +1216,7 @@ def concat(bags):
     >>> list(c)
     [1, 2, 3, 4, 5, 6]
     """
-    name = next(names)
+    name = 'concat-' + tokenize(*bags)
     counter = itertools.count(0)
     dsk = dict(((name, next(counter)), key)
                for bag in bags for key in sorted(bag._keys()))


Test patch:
diff --git a/dask/bag/tests/test_bag.py b/dask/bag/tests/test_bag.py
index 6b0273e34..6f46f4434 100644
--- a/dask/bag/tests/test_bag.py
+++ b/dask/bag/tests/test_bag.py
@@ -59,6 +59,7 @@ def test_map():
     expected = merge(dsk, dict(((c.name, i), (reify, (map, inc, (b.name, i))))
                                for i in range(b.npartitions)))
     assert c.dask == expected
+    assert c.name == b.map(inc).name
 
 
 def test_map_function_with_multiple_arguments():
@@ -108,10 +109,14 @@ def test_filter():
                                 (reify, (filter, iseven, (b.name, i))))
                                for i in range(b.npartitions)))
     assert c.dask == expected
+    assert c.name == b.filter(iseven).name
 
 
 def test_remove():
-    assert list(b.remove(lambda x: x % 2 == 0)) == [1, 3] * 3
+    f = lambda x: x % 2 == 0
+    c = b.remove(f)
+    assert list(c) == [1, 3] * 3
+    assert c.name == b.remove(f).name
 
 
 def test_iter():
@@ -126,17 +131,26 @@ def test_pluck():
     assert set(b.pluck(0)) == set([1, 2, 3, 4])
     assert set(b.pluck(1)) == set([10, 20, 30, 40])
     assert set(b.pluck([1, 0])) == set([(10, 1), (20, 2), (30, 3), (40, 4)])
+    assert b.pluck([1, 0]).name == b.pluck([1, 0]).name
 
 
 def test_pluck_with_default():
     b = db.from_sequence(['Hello', '', 'World'])
     assert raises(IndexError, lambda: list(b.pluck(0)))
     assert list(b.pluck(0, None)) == ['H', None, 'W']
+    assert b.pluck(0, None).name == b.pluck(0, None).name
+    assert b.pluck(0).name != b.pluck(0, None).name
 
 
 def test_fold():
-    assert b.fold(add).compute() == sum(L)
-    assert b.fold(add, initial=10).compute() == sum(L) + 10 * b.npartitions
+    c = b.fold(add)
+    assert c.compute() == sum(L)
+    assert c.key == b.fold(add).key
+
+    c2 = b.fold(add, initial=10)
+    assert c2.key != c.key
+    assert c2.compute() == sum(L) + 10 * b.npartitions
+    assert c2.key == b.fold(add, initial=10).key
 
     c = db.from_sequence(range(5), npartitions=3)
     def binop(acc, x):
@@ -144,7 +158,9 @@ def test_fold():
         acc.add(x)
         return acc
 
-    assert c.fold(binop, set.union, initial=set()).compute() == set(c)
+    d = c.fold(binop, set.union, initial=set())
+    assert d.compute() == set(c)
+    assert d.key == c.fold(binop, set.union, initial=set()).key
 
     d = db.from_sequence('hello')
     assert set(d.fold(lambda a, b: ''.join([a, b]), initial='').compute()) == set('hello')
@@ -156,23 +172,31 @@ def test_fold():
 
 def test_distinct():
     assert sorted(b.distinct()) == [0, 1, 2, 3, 4]
+    assert b.distinct().name == b.distinct().name
 
 
 def test_frequencies():
-    assert dict(b.frequencies()) == {0: 3, 1: 3, 2: 3, 3: 3, 4: 3}
-    assert dict(b.frequencies(split_every=2)) == {0: 3, 1: 3, 2: 3, 3: 3, 4: 3}
+    c = b.frequencies()
+    assert dict(c) == {0: 3, 1: 3, 2: 3, 3: 3, 4: 3}
+    c2 = b.frequencies(split_every=2)
+    assert dict(c2) == {0: 3, 1: 3, 2: 3, 3: 3, 4: 3}
+    assert c.name == b.frequencies().name
+    assert c.name != c2.name
+    assert c2.name == b.frequencies(split_every=2).name
 
 
 def test_topk():
     assert list(b.topk(4)) == [4, 4, 4, 3]
     assert list(b.topk(4, key=lambda x: -x).compute(get=dask.get)) == \
             [0, 0, 0, 1]
+    assert b.topk(4).name == b.topk(4).name
 
 
 def test_topk_with_non_callable_key():
     b = db.from_sequence([(1, 10), (2, 9), (3, 8)], npartitions=2)
     assert list(b.topk(2, key=1)) == [(1, 10), (2, 9)]
     assert list(b.topk(2, key=0)) == [(3, 8), (2, 9)]
+    assert b.topk(2, key=1).name == b.topk(2, key=1).name
 
 
 def test_topk_with_multiarg_lambda():
@@ -183,6 +207,7 @@ def test_topk_with_multiarg_lambda():
 def test_lambdas():
     assert list(b.map(lambda x: x + 1)) == list(b.map(inc))
 
+
 def test_reductions():
     assert int(b.count()) == 15
     assert int(b.sum()) == 30
@@ -190,6 +215,8 @@ def test_reductions():
     assert int(b.min()) == 0
     assert int(b.any()) == True
     assert int(b.all()) == False  # some zeros exist
+    assert b.all().key == b.all().key
+    assert b.all().key != b.any().key
 
 
 def test_tree_reductions():
@@ -208,30 +235,39 @@ def test_tree_reductions():
     assert c.compute() == d.compute()
     assert len(c.dask) > len(d.dask)
 
+    assert c.key != d.key
+    assert c.key == b.sum(split_every=2).key
+    assert c.key != b.sum().key
+
 
 def test_mean():
     assert b.mean().compute(get=dask.get) == 2.0
     assert float(b.mean()) == 2.0
 
+
 def test_std():
     assert b.std().compute(get=dask.get) == math.sqrt(2.0)
     assert float(b.std()) == math.sqrt(2.0)
 
+
 def test_var():
     assert b.var().compute(get=dask.get) == 2.0
     assert float(b.var()) == 2.0
 
 
 def test_join():
-    assert list(b.join([1, 2, 3], on_self=isodd, on_other=iseven)) == \
-            list(join(iseven, [1, 2, 3], isodd, list(b)))
+    c = b.join([1, 2, 3], on_self=isodd, on_other=iseven)
+    assert list(c) == list(join(iseven, [1, 2, 3], isodd, list(b)))
     assert list(b.join([1, 2, 3], isodd)) == \
             list(join(isodd, [1, 2, 3], isodd, list(b)))
+    assert c.name == b.join([1, 2, 3], on_self=isodd, on_other=iseven).name
+
 
 def test_foldby():
     c = b.foldby(iseven, add, 0, add, 0)
     assert (reduceby, iseven, add, (b.name, 0), 0) in list(c.dask.values())
     assert set(c) == set(reduceby(iseven, lambda acc, x: acc + x, L, 0).items())
+    assert c.name == b.foldby(iseven, add, 0, add, 0).name
 
     c = b.foldby(iseven, lambda acc, x: acc + x)
     assert set(c) == set(reduceby(iseven, lambda acc, x: acc + x, L, 0).items())
@@ -239,6 +275,8 @@ def test_foldby():
 
 def test_map_partitions():
     assert list(b.map_partitions(len)) == [5, 5, 5]
+    assert b.map_partitions(len).name == b.map_partitions(len).name
+    assert b.map_partitions(lambda a: len(a) + 1).name != b.map_partitions(len).name
 
 
 def test_lazify_task():
@@ -296,6 +334,7 @@ def test_map_is_lazy():
     from dask.bag.core import map
     assert isinstance(map(lambda x: x, [1, 2, 3]), Iterator)
 
+
 def test_can_use_dict_to_make_concrete():
     assert isinstance(dict(b.frequencies()), dict)
 
@@ -318,6 +357,8 @@ def test_from_castra():
                 list(default) == [(i, str(i)) for i in range(100)])
         assert list(with_columns) == list(range(100))
         assert list(with_index) == list(zip(range(100), range(100)))
+        assert default.name != with_columns.name != with_index.name
+        assert with_index.name == db.from_castra(c, 'x', index=True).name
 
 
 @pytest.mark.slow
@@ -477,6 +518,8 @@ def test_product():
     z = x.product(y)
     assert set(z) == set([(i, j) for i in [1, 2, 3, 4] for j in [10, 20, 30]])
 
+    assert z.name != b2.name
+    assert z.name == x.product(y).name
 
 
 def test_partition_collect():
@@ -491,14 +534,16 @@ def test_partition_collect():
 
 
 def test_groupby():
-    c = b.groupby(lambda x: x)
+    c = b.groupby(identity)
     result = dict(c)
     assert result == {0: [0, 0 ,0],
                       1: [1, 1, 1],
                       2: [2, 2, 2],
                       3: [3, 3, 3],
                       4: [4, 4, 4]}
-    assert b.groupby(lambda x: x).npartitions == b.npartitions
+    assert c.npartitions == b.npartitions
+    assert c.name == b.groupby(identity).name
+    assert c.name != b.groupby(lambda x: x + 1).name
 
 
 def test_groupby_with_indexer():
@@ -507,6 +552,7 @@ def test_groupby_with_indexer():
     assert valmap(sorted, result) == {1: [[1, 2, 3], [1, 4, 9]],
                                       2: [[2, 3, 4]]}
 
+
 def test_groupby_with_npartitions_changed():
     result = b.groupby(lambda x: x, npartitions=1)
     result2 = dict(result)
@@ -525,6 +571,10 @@ def test_concat():
     c = db.concat([a, b])
     assert list(c) == [1, 2, 3, 4, 5, 6]
 
+    assert c.name == db.concat([a, b]).name
+    assert b.concat().name != a.concat().name
+    assert b.concat().name == b.concat().name
+
     b = db.from_sequence([1, 2, 3]).map(lambda x: x * [1, 2, 3])
     assert list(b.concat()) == [1, 2, 3] * sum([1, 2, 3])
 
@@ -570,6 +620,10 @@ def test_to_dataframe():
 
     assert (df2.compute().values == df.compute().values).all()
 
+    assert df2._name == b.to_dataframe()._name
+    assert df2._name != df._name
+
+
 def test_to_textfiles():
     b = db.from_sequence(['abc', '123', 'xyz'], npartitions=2)
     dir = mkdtemp()
@@ -648,6 +702,8 @@ def test_string_namespace():
     assert list(b.str.match('*Smith')) == ['Alice Smith', 'Charlie Smith']
 
     assert raises(AttributeError, lambda: b.str.sfohsofhf)
+    assert b.str.match('*Smith').name == b.str.match('*Smith').name
+    assert b.str.match('*Smith').name != b.str.match('*John').name
 
 
 def test_string_namespace_with_unicode():
@@ -743,6 +799,7 @@ def test_from_imperative():
     from dask.imperative import value
     a, b, c = value([1, 2, 3]), value([4, 5, 6]), value([7, 8, 9])
     bb = from_imperative([a, b, c])
+    assert bb.name == from_imperative([a, b, c]).name
 
     assert isinstance(bb, Bag)
     assert list(bb) == [1, 2, 3, 4, 5, 6, 7, 8, 9]