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()))

dask.bag does not use hashed keys
We should use `tokenize` rather than the current `tokens` within dask.bag

cc @jcrist 
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]

pip install -e .[complete]
parse_log_pytest
numpy>=1.16.0 pandas>=1.0.0 cloudpickle partd distributed s3fs toolz psutil pytables bokeh bcolz scipy h5py ipython
pytest
apt-get update apt-get install -y graphviz liblzma-dev
3.5
pytest --no-header -rA --tb=line --color=no -p no:cacheprovider -W ignore::DeprecationWarning
aiobotocore @ file:///opt/conda/conda-bld/aiobotocore_1643638228694/work
aiohttp @ file:///tmp/build/80754af9/aiohttp_1632748060317/work
aioitertools @ file:///tmp/build/80754af9/aioitertools_1607109665762/work
async-timeout==3.0.1
attrs @ file:///opt/conda/conda-bld/attrs_1642510447205/work
backcall @ file:///home/ktietz/src/ci/backcall_1611930011877/work
bcolz==1.2.1
bokeh @ file:///tmp/build/80754af9/bokeh_1620710048147/work
botocore @ file:///opt/conda/conda-bld/botocore_1642672735464/work
brotlipy==0.7.0
certifi==2021.5.30
cffi @ file:///tmp/build/80754af9/cffi_1625814693874/work
chardet @ file:///tmp/build/80754af9/chardet_1607706739153/work
click==8.0.3
cloudpickle @ file:///tmp/build/80754af9/cloudpickle_1632508026186/work
contextvars==2.4
cryptography @ file:///tmp/build/80754af9/cryptography_1635366128178/work
cytoolz==0.11.0
-e git+https://github.com/dask/dask.git@d82cf2ac3fa3a61912b7934afe7b2fe9e14cc4ff#egg=dask
decorator @ file:///opt/conda/conda-bld/decorator_1643638310831/work
distributed @ file:///tmp/build/80754af9/distributed_1615054599257/work
fsspec @ file:///opt/conda/conda-bld/fsspec_1642510437511/work
h5py==2.10.0
HeapDict @ file:///Users/ktietz/demo/mc3/conda-bld/heapdict_1630598515714/work
idna @ file:///tmp/build/80754af9/idna_1637925883363/work
idna-ssl @ file:///tmp/build/80754af9/idna_ssl_1611752490495/work
immutables @ file:///tmp/build/80754af9/immutables_1628888996840/work
importlib-metadata==4.8.3
iniconfig==1.1.1
ipython @ file:///tmp/build/80754af9/ipython_1593447367857/work
ipython-genutils @ file:///tmp/build/80754af9/ipython_genutils_1606773439826/work
jedi @ file:///tmp/build/80754af9/jedi_1606932572482/work
Jinja2 @ file:///opt/conda/conda-bld/jinja2_1647436528585/work
jmespath @ file:///Users/ktietz/demo/mc3/conda-bld/jmespath_1630583964805/work
locket==0.2.1
MarkupSafe @ file:///tmp/build/80754af9/markupsafe_1621528150516/work
mock @ file:///tmp/build/80754af9/mock_1607622725907/work
msgpack @ file:///tmp/build/80754af9/msgpack-python_1612287171716/work
multidict @ file:///tmp/build/80754af9/multidict_1607367768400/work
numexpr @ file:///tmp/build/80754af9/numexpr_1618853194344/work
numpy @ file:///tmp/build/80754af9/numpy_and_numpy_base_1603483703303/work
olefile @ file:///Users/ktietz/demo/mc3/conda-bld/olefile_1629805411829/work
packaging @ file:///tmp/build/80754af9/packaging_1637314298585/work
pandas==1.1.5
parso==0.7.0
partd @ file:///opt/conda/conda-bld/partd_1647245470509/work
pexpect @ file:///tmp/build/80754af9/pexpect_1605563209008/work
pickleshare @ file:///tmp/build/80754af9/pickleshare_1606932040724/work
Pillow @ file:///tmp/build/80754af9/pillow_1625670622947/work
pluggy==1.0.0
prompt-toolkit @ file:///tmp/build/80754af9/prompt-toolkit_1633440160888/work
psutil @ file:///tmp/build/80754af9/psutil_1612297621795/work
ptyprocess @ file:///tmp/build/80754af9/ptyprocess_1609355006118/work/dist/ptyprocess-0.7.0-py2.py3-none-any.whl
py==1.11.0
pycparser @ file:///tmp/build/80754af9/pycparser_1636541352034/work
Pygments @ file:///opt/conda/conda-bld/pygments_1644249106324/work
pyOpenSSL @ file:///opt/conda/conda-bld/pyopenssl_1643788558760/work
pyparsing @ file:///tmp/build/80754af9/pyparsing_1635766073266/work
PySocks @ file:///tmp/build/80754af9/pysocks_1605305763431/work
pytest==7.0.1
python-dateutil @ file:///tmp/build/80754af9/python-dateutil_1626374649649/work
pytz==2021.3
PyYAML==5.4.1
s3fs @ file:///opt/conda/conda-bld/s3fs_1643701468749/work
scipy @ file:///tmp/build/80754af9/scipy_1597686635649/work
six @ file:///tmp/build/80754af9/six_1644875935023/work
sortedcontainers @ file:///tmp/build/80754af9/sortedcontainers_1623949099177/work
tables==3.6.1
tblib @ file:///Users/ktietz/demo/mc3/conda-bld/tblib_1629402031467/work
tomli==1.2.3
toolz @ file:///tmp/build/80754af9/toolz_1636545406491/work
tornado @ file:///tmp/build/80754af9/tornado_1606942266872/work
traitlets @ file:///tmp/build/80754af9/traitlets_1632746497744/work
typing_extensions @ file:///opt/conda/conda-bld/typing_extensions_1647553014482/work
urllib3 @ file:///opt/conda/conda-bld/urllib3_1643638302206/work
wcwidth @ file:///Users/ktietz/demo/mc3/conda-bld/wcwidth_1629357192024/work
wrapt==1.12.1
yarl @ file:///tmp/build/80754af9/yarl_1606939915466/work
zict==2.0.0
zipp==3.6.0

name: dask
channels:
  - defaults
  - https://repo.anaconda.com/pkgs/main
  - https://repo.anaconda.com/pkgs/r
  - conda-forge
dependencies:
  - _libgcc_mutex=0.1=main
  - _openmp_mutex=5.1=1_gnu
  - aiobotocore=2.1.0=pyhd3eb1b0_0
  - aiohttp=3.7.4.post0=py36h7f8727e_2
  - aioitertools=0.7.1=pyhd3eb1b0_0
  - async-timeout=3.0.1=py36h06a4308_0
  - attrs=21.4.0=pyhd3eb1b0_0
  - backcall=0.2.0=pyhd3eb1b0_0
  - bcolz=1.2.1=py36h04863e7_0
  - blas=1.0=openblas
  - blosc=1.21.3=h6a678d5_0
  - bokeh=2.3.2=py36h06a4308_0
  - botocore=1.23.24=pyhd3eb1b0_0
  - brotlipy=0.7.0=py36h27cfd23_1003
  - bzip2=1.0.8=h5eee18b_6
  - ca-certificates=2025.2.25=h06a4308_0
  - certifi=2021.5.30=py36h06a4308_0
  - cffi=1.14.6=py36h400218f_0
  - chardet=4.0.0=py36h06a4308_1003
  - click=8.0.3=pyhd3eb1b0_0
  - cloudpickle=2.0.0=pyhd3eb1b0_0
  - contextvars=2.4=py_0
  - cryptography=35.0.0=py36hd23ed53_0
  - cytoolz=0.11.0=py36h7b6447c_0
  - decorator=5.1.1=pyhd3eb1b0_0
  - distributed=2021.3.0=py36h06a4308_0
  - freetype=2.12.1=h4a9f257_0
  - fsspec=2022.1.0=pyhd3eb1b0_0
  - giflib=5.2.2=h5eee18b_0
  - h5py=2.10.0=py36h7918eee_0
  - hdf5=1.10.4=hb1b8bf9_0
  - heapdict=1.0.1=pyhd3eb1b0_0
  - idna=3.3=pyhd3eb1b0_0
  - idna_ssl=1.1.0=py36h06a4308_0
  - immutables=0.16=py36h7f8727e_0
  - ipython=7.16.1=py36h5ca1d4c_0
  - ipython_genutils=0.2.0=pyhd3eb1b0_1
  - jedi=0.17.2=py36h06a4308_1
  - jinja2=3.0.3=pyhd3eb1b0_0
  - jmespath=0.10.0=pyhd3eb1b0_0
  - jpeg=9e=h5eee18b_3
  - lcms2=2.16=hb9589c4_0
  - ld_impl_linux-64=2.40=h12ee557_0
  - lerc=4.0.0=h6a678d5_0
  - libdeflate=1.22=h5eee18b_0
  - libffi=3.3=he6710b0_2
  - libgcc-ng=11.2.0=h1234567_1
  - libgfortran-ng=7.5.0=ha8ba4b0_17
  - libgfortran4=7.5.0=ha8ba4b0_17
  - libgomp=11.2.0=h1234567_1
  - libopenblas=0.3.18=hf726d26_0
  - libpng=1.6.39=h5eee18b_0
  - libstdcxx-ng=11.2.0=h1234567_1
  - libtiff=4.5.1=hffd6297_1
  - libwebp=1.2.4=h11a3e52_1
  - libwebp-base=1.2.4=h5eee18b_1
  - locket=0.2.1=py36h06a4308_1
  - lz4-c=1.9.4=h6a678d5_1
  - lzo=2.10=h7b6447c_2
  - markupsafe=2.0.1=py36h27cfd23_0
  - mock=4.0.3=pyhd3eb1b0_0
  - msgpack-python=1.0.2=py36hff7bd54_1
  - multidict=5.1.0=py36h27cfd23_2
  - ncurses=6.4=h6a678d5_0
  - numexpr=2.7.3=py36h4be448d_1
  - numpy=1.19.2=py36h6163131_0
  - numpy-base=1.19.2=py36h75fe3a5_0
  - olefile=0.46=pyhd3eb1b0_0
  - openssl=1.1.1w=h7f8727e_0
  - packaging=21.3=pyhd3eb1b0_0
  - pandas=1.1.5=py36ha9443f7_0
  - parso=0.7.0=py_0
  - partd=1.2.0=pyhd3eb1b0_1
  - pexpect=4.8.0=pyhd3eb1b0_3
  - pickleshare=0.7.5=pyhd3eb1b0_1003
  - pillow=8.3.1=py36h5aabda8_0
  - pip=21.2.2=py36h06a4308_0
  - prompt-toolkit=3.0.20=pyhd3eb1b0_0
  - psutil=5.8.0=py36h27cfd23_1
  - ptyprocess=0.7.0=pyhd3eb1b0_2
  - pycparser=2.21=pyhd3eb1b0_0
  - pygments=2.11.2=pyhd3eb1b0_0
  - pyopenssl=22.0.0=pyhd3eb1b0_0
  - pyparsing=3.0.4=pyhd3eb1b0_0
  - pysocks=1.7.1=py36h06a4308_0
  - pytables=3.6.1=py36h71ec239_0
  - python=3.6.13=h12debd9_1
  - python-dateutil=2.8.2=pyhd3eb1b0_0
  - pytz=2021.3=pyhd3eb1b0_0
  - pyyaml=5.4.1=py36h27cfd23_1
  - readline=8.2=h5eee18b_0
  - s3fs=2022.1.0=pyhd3eb1b0_0
  - scipy=1.5.2=py36habc2bb6_0
  - setuptools=58.0.4=py36h06a4308_0
  - six=1.16.0=pyhd3eb1b0_1
  - sortedcontainers=2.4.0=pyhd3eb1b0_0
  - sqlite=3.45.3=h5eee18b_0
  - tblib=1.7.0=pyhd3eb1b0_0
  - tk=8.6.14=h39e8969_0
  - toolz=0.11.2=pyhd3eb1b0_0
  - tornado=6.1=py36h27cfd23_0
  - traitlets=4.3.3=py36h06a4308_0
  - typing-extensions=4.1.1=hd3eb1b0_0
  - typing_extensions=4.1.1=pyh06a4308_0
  - urllib3=1.26.8=pyhd3eb1b0_0
  - wcwidth=0.2.5=pyhd3eb1b0_0
  - wheel=0.37.1=pyhd3eb1b0_0
  - wrapt=1.12.1=py36h7b6447c_1
  - xz=5.6.4=h5eee18b_1
  - yaml=0.2.5=h7b6447c_0
  - yarl=1.6.3=py36h27cfd23_0
  - zict=2.0.0=pyhd3eb1b0_0
  - zlib=1.2.13=h5eee18b_1
  - zstd=1.5.6=hc292b87_0
  - pip:
      - importlib-metadata==4.8.3
      - iniconfig==1.1.1
      - pluggy==1.0.0
      - py==1.11.0
      - pytest==7.0.1
      - tomli==1.2.3
      - zipp==3.6.0
prefix: /opt/conda/envs/dask

dask/bag/tests/test_bag.py::test_map dask/bag/tests/test_bag.py::test_filter dask/bag/tests/test_bag.py::test_remove dask/bag/tests/test_bag.py::test_pluck dask/bag/tests/test_bag.py::test_pluck_with_default dask/bag/tests/test_bag.py::test_fold dask/bag/tests/test_bag.py::test_distinct dask/bag/tests/test_bag.py::test_frequencies dask/bag/tests/test_bag.py::test_topk dask/bag/tests/test_bag.py::test_topk_with_non_callable_key dask/bag/tests/test_bag.py::test_reductions dask/bag/tests/test_bag.py::test_tree_reductions dask/bag/tests/test_bag.py::test_join dask/bag/tests/test_bag.py::test_foldby dask/bag/tests/test_bag.py::test_map_partitions dask/bag/tests/test_bag.py::test_product dask/bag/tests/test_bag.py::test_groupby dask/bag/tests/test_bag.py::test_concat dask/bag/tests/test_bag.py::test_string_namespace
dask/bag/tests/test_bag.py::test_Bag dask/bag/tests/test_bag.py::test_keys dask/bag/tests/test_bag.py::test_map_function_with_multiple_arguments dask/bag/tests/test_bag.py::test_map_with_constructors dask/bag/tests/test_bag.py::test_map_with_builtins dask/bag/tests/test_bag.py::test_iter dask/bag/tests/test_bag.py::test_topk_with_multiarg_lambda dask/bag/tests/test_bag.py::test_lambdas dask/bag/tests/test_bag.py::test_mean dask/bag/tests/test_bag.py::test_std dask/bag/tests/test_bag.py::test_var dask/bag/tests/test_bag.py::test_lazify_task dask/bag/tests/test_bag.py::test_lazify dask/bag/tests/test_bag.py::test_inline_singleton_lists dask/bag/tests/test_bag.py::test_take dask/bag/tests/test_bag.py::test_map_is_lazy dask/bag/tests/test_bag.py::test_can_use_dict_to_make_concrete dask/bag/tests/test_bag.py::test_from_filenames dask/bag/tests/test_bag.py::test_from_filenames_gzip dask/bag/tests/test_bag.py::test_from_filenames_bz2 dask/bag/tests/test_bag.py::test_from_filenames_large dask/bag/tests/test_bag.py::test_from_filenames_encoding dask/bag/tests/test_bag.py::test_from_filenames_large_gzip dask/bag/tests/test_bag.py::test__parse_s3_URI dask/bag/tests/test_bag.py::test_from_sequence dask/bag/tests/test_bag.py::test_from_long_sequence dask/bag/tests/test_bag.py::test_partition_collect dask/bag/tests/test_bag.py::test_groupby_with_indexer dask/bag/tests/test_bag.py::test_groupby_with_npartitions_changed dask/bag/tests/test_bag.py::test_concat_after_map dask/bag/tests/test_bag.py::test_args dask/bag/tests/test_bag.py::test_to_dataframe dask/bag/tests/test_bag.py::test_to_textfiles dask/bag/tests/test_bag.py::test_to_textfiles_encoding dask/bag/tests/test_bag.py::test_to_textfiles_inputs dask/bag/tests/test_bag.py::test_bz2_stream dask/bag/tests/test_bag.py::test_string_namespace_with_unicode dask/bag/tests/test_bag.py::test_str_empty_split dask/bag/tests/test_bag.py::test_stream_decompress dask/bag/tests/test_bag.py::test_map_with_iterator_function dask/bag/tests/test_bag.py::test_ensure_compute_output_is_concrete dask/bag/tests/test_bag.py::test_bag_class_extend dask/bag/tests/test_bag.py::test_gh715 dask/bag/tests/test_bag.py::test_bag_compute_forward_kwargs dask/bag/tests/test_bag.py::test_to_imperative dask/bag/tests/test_bag.py::test_from_imperative
dask/bag/core.py
dask/bag/core.py