Repository: dask/dask

Issue:
Groupby nunique on empty dataframe fails
There seems to be an issue when calling Groupby.nunique on an empty dataframe

On plain pandas:
```python
ua_pd_df = pandas.DataFrame(columns=['username', 'user_agent', 'requests'])
ua_pd_df.groupby(['user_agent']).username.nunique()
Series([], Name: username, dtype: int64)
```

On dask dataframe:

```python
ua_dd = dd.from_pandas(ua_pd_df, npartitions=5) 
ua_dd.groupby('user_agent').username.nunique().compute() 

---------------------------------------------------------------------------
KeyError                                  Traceback (most recent call last)
<ipython-input-76-d6bfb5ac8c8c> in <module>
----> 1 ua_df[['username', 'user_agent', 'requests']].groupby('user_agent').username.nunique().compute()

~/virtualenvs/esa-engine-dask3/lib/python3.6/site-packages/dask/base.py in compute(self, **kwargs)
    154         dask.base.compute
    155         """
--> 156         (result,) = compute(self, traverse=False, **kwargs)
    157         return result
    158 

~/virtualenvs/esa-engine-dask3/lib/python3.6/site-packages/dask/base.py in compute(*args, **kwargs)
    396     keys = [x.__dask_keys__() for x in collections]
    397     postcomputes = [x.__dask_postcompute__() for x in collections]
--> 398     results = schedule(dsk, keys, **kwargs)
    399     return repack([f(r, *a) for r, (f, a) in zip(results, postcomputes)])
    400 

~/virtualenvs/esa-engine-dask3/lib/python3.6/site-packages/distributed/client.py in get(self, dsk, keys, restrictions, loose_restrictions, resources, sync, asynchronous, direct, retries, priority, fifo_timeout, actors, **kwargs)
   2329             try:
   2330                 results = self.gather(packed, asynchronous=asynchronous,
-> 2331                                       direct=direct)
   2332             finally:
   2333                 for f in futures.values():

~/virtualenvs/esa-engine-dask3/lib/python3.6/site-packages/distributed/client.py in gather(self, futures, errors, maxsize, direct, asynchronous)
   1653             return self.sync(self._gather, futures, errors=errors,
   1654                              direct=direct, local_worker=local_worker,
-> 1655                              asynchronous=asynchronous)
   1656 
   1657     @gen.coroutine

~/virtualenvs/esa-engine-dask3/lib/python3.6/site-packages/distributed/client.py in sync(self, func, *args, **kwargs)
    673             return future
    674         else:
--> 675             return sync(self.loop, func, *args, **kwargs)
    676 
    677     def __repr__(self):

~/virtualenvs/esa-engine-dask3/lib/python3.6/site-packages/distributed/utils.py in sync(loop, func, *args, **kwargs)
    275             e.wait(10)
    276     if error[0]:
--> 277         six.reraise(*error[0])
    278     else:
    279         return result[0]

~/virtualenvs/esa-engine-dask3/lib/python3.6/site-packages/six.py in reraise(tp, value, tb)
    691             if value.__traceback__ is not tb:
    692                 raise value.with_traceback(tb)
--> 693             raise value
    694         finally:
    695             value = None

~/virtualenvs/esa-engine-dask3/lib/python3.6/site-packages/distributed/utils.py in f()
    260             if timeout is not None:
    261                 future = gen.with_timeout(timedelta(seconds=timeout), future)
--> 262             result[0] = yield future
    263         except Exception as exc:
    264             error[0] = sys.exc_info()

~/virtualenvs/esa-engine-dask3/lib/python3.6/site-packages/tornado/gen.py in run(self)
   1131 
   1132                     try:
-> 1133                         value = future.result()
   1134                     except Exception:
   1135                         self.had_exception = True

~/virtualenvs/esa-engine-dask3/lib/python3.6/site-packages/tornado/gen.py in run(self)
   1139                     if exc_info is not None:
   1140                         try:
-> 1141                             yielded = self.gen.throw(*exc_info)
   1142                         finally:
   1143                             # Break up a reference to itself

~/virtualenvs/esa-engine-dask3/lib/python3.6/site-packages/distributed/client.py in _gather(self, futures, errors, direct, local_worker)
   1494                             six.reraise(type(exception),
   1495                                         exception,
-> 1496                                         traceback)
   1497                     if errors == 'skip':
   1498                         bad_keys.add(key)

~/virtualenvs/esa-engine-dask3/lib/python3.6/site-packages/six.py in reraise(tp, value, tb)
    690                 value = tp()
    691             if value.__traceback__ is not tb:
--> 692                 raise value.with_traceback(tb)
    693             raise value
    694         finally:

~/virtualenvs/esa-engine-dask3/lib/python3.6/site-packages/dask/compatibility.py in apply()
     91     def apply(func, args, kwargs=None):
     92         if kwargs:
---> 93             return func(*args, **kwargs)
     94         else:
     95             return func(*args)

~/virtualenvs/esa-engine-dask3/lib/python3.6/site-packages/dask/dataframe/groupby.py in _nunique_df_aggregate()
    315 
    316 def _nunique_df_aggregate(df, levels, name):
--> 317     return df.groupby(level=levels, sort=False)[name].nunique()
    318 
    319 

~/virtualenvs/esa-engine-dask3/lib/python3.6/site-packages/pandas/core/base.py in __getitem__()
    273         else:
    274             if key not in self.obj:
--> 275                 raise KeyError("Column not found: {key}".format(key=key))
    276             return self._gotitem(key, ndim=1)
    277 

KeyError: 'Column not found: username'
```
Any other method to use to get the number of unique values per group?

Dask: v1.1.1
Python: v3.6
Dask distributed: v1.25.3

Discussion:
jcrist: Looks like empty groupby-reductions fail for old versions of pandas (at least up to 0.20.2). I'm fine punting on this issue and just skipping that test for older pandas versions (things work fine on recent releases). Thoughts @TomAugspurger?
TomAugspurger: Yep, skipping for old pandas seems fine.

On Tue, Feb 19, 2019 at 9:48 AM Jim Crist <notifications@github.com> wrote:

> Looks like empty groupby-reductions fail for old versions of pandas (at
> least up to 0.20.2). I'm fine punting on this issue and just skipping that
> test for older pandas versions (things work fine on recent releases).
> Thoughts @TomAugspurger <https://github.com/TomAugspurger>?
>
> —
> You are receiving this because you were mentioned.
> Reply to this email directly, view it on GitHub
> <https://github.com/dask/dask/pull/4504#issuecomment-465186424>, or mute
> the thread
> <https://github.com/notifications/unsubscribe-auth/ABQHInLUARsKXKmJ0i2qq6uub1g595u8ks5vPBzBgaJpZM4bDK55>
> .
>


Patch:
diff --git a/dask/array/core.py b/dask/array/core.py
index 61158f7c5..08ca79f7e 100644
--- a/dask/array/core.py
+++ b/dask/array/core.py
@@ -2428,7 +2428,7 @@ def unify_chunks(*args, **kwargs):
     if not args:
         return {}, []
 
-    arginds = [(asarray(a) if ind is not None else a, ind)
+    arginds = [(asanyarray(a) if ind is not None else a, ind)
                for a, ind in partition(2, args)]  # [x, ij, y, jk]
     args = list(concat(arginds))  # [(x, ij), (y, jk)]
     warn = kwargs.get('warn', True)
@@ -2934,19 +2934,12 @@ def asarray(a, **kwargs):
     >>> da.asarray(y)
     dask.array<array, shape=(2, 3), dtype=int64, chunksize=(2, 3)>
     """
-    def frame_types():
-        try:
-            import dask.dataframe as dd
-            return (dd.Series, dd.DataFrame)
-        except ImportError:
-            return ()
-
     if isinstance(a, Array):
         return a
-    if isinstance(a, (list, tuple)) and any(isinstance(i, Array) for i in a):
-        a = stack(a)
-    elif isinstance(a, frame_types()):
+    elif hasattr(a, 'to_dask_array'):
         return a.to_dask_array()
+    elif isinstance(a, (list, tuple)) and any(isinstance(i, Array) for i in a):
+        a = stack(a)
     elif not isinstance(getattr(a, 'shape', None), Iterable):
         a = np.asarray(a)
     return from_array(a, chunks=a.shape, getitem=getter_inline, **kwargs)
diff --git a/dask/bag/core.py b/dask/bag/core.py
index 141ade7e2..16551d3f6 100644
--- a/dask/bag/core.py
+++ b/dask/bag/core.py
@@ -343,7 +343,7 @@ class Item(DaskMethodsMixin):
         self.dask, self.key = state
 
     def apply(self, func):
-        name = 'apply-{0}-{1}'.format(funcname(func), tokenize(self, func))
+        name = '{0}-{1}'.format(funcname(func), tokenize(self, func, 'apply'))
         dsk = {name: (func, self.key)}
         graph = HighLevelGraph.from_collections(name, dsk, dependencies=[self])
         return Item(graph, name)
@@ -538,8 +538,8 @@ class Bag(DaskMethodsMixin):
         >>> b.starmap(myadd, z=max_second).compute()
         [13, 17, 21, 25, 29]
         """
-        name = 'starmap-{0}-{1}'.format(funcname(func),
-                                        tokenize(self, func, kwargs))
+        name = '{0}-{1}'.format(funcname(func),
+                                tokenize(self, func, 'starmap', **kwargs))
         dependencies = [self]
         if kwargs:
             kwargs, collections = unpack_scalar_dask_kwargs(kwargs)
@@ -1849,7 +1849,7 @@ def bag_map(func, *args, **kwargs):
     >>> db.map(myadd, b, b.max()).compute()
     [4, 5, 6, 7, 8]
     """
-    name = 'map-%s-%s' % (funcname(func), tokenize(func, args, kwargs))
+    name = '%s-%s' % (funcname(func), tokenize(func, 'map', *args, **kwargs))
     dsk = {}
     dependencies = []
 
@@ -1942,8 +1942,8 @@ def map_partitions(func, *args, **kwargs):
     single graph, and then computes everything at once, and in some cases
     may be more efficient.
     """
-    name = 'map-partitions-%s-%s' % (funcname(func),
-                                     tokenize(func, args, kwargs))
+    name = '%s-%s' % (funcname(func),
+                      tokenize(func, 'map-partitions', *args, **kwargs))
     dsk = {}
     dependencies = []
 
diff --git a/dask/dataframe/core.py b/dask/dataframe/core.py
index 0245a4ca7..bcff938af 100644
--- a/dask/dataframe/core.py
+++ b/dask/dataframe/core.py
@@ -27,7 +27,7 @@ from ..compatibility import (apply, operator_div, bind_method, string_types,
 from ..context import globalmethod
 from ..utils import (random_state_data, pseudorandom, derived_from, funcname,
                      memory_repr, put_lines, M, key_split, OperatorMethodMixin,
-                     is_arraylike)
+                     is_arraylike, typename)
 from ..array.core import Array, normalize_arg
 from ..blockwise import blockwise, Blockwise
 from ..base import DaskMethodsMixin, tokenize, dont_optimize, is_dask_collection
@@ -90,7 +90,7 @@ class Scalar(DaskMethodsMixin, OperatorMethodMixin):
         meta = make_meta(meta)
         if is_dataframe_like(meta) or is_series_like(meta) or is_index_like(meta):
             raise TypeError("Expected meta to specify scalar, got "
-                            "{0}".format(type(meta).__name__))
+                            "{0}".format(typename(type(meta))))
         self._meta = meta
 
     def __dask_graph__(self):
@@ -253,8 +253,8 @@ class _Frame(DaskMethodsMixin, OperatorMethodMixin):
         meta = make_meta(meta)
         if not isinstance(meta, self._partition_type):
             raise TypeError("Expected meta to specify type {0}, got type "
-                            "{1}".format(self._partition_type.__name__,
-                                         type(meta).__name__))
+                            "{1}".format(typename(self._partition_type),
+                                         typename(type(meta))))
         self._meta = meta
         self.divisions = tuple(divisions)
 
@@ -2610,7 +2610,7 @@ class DataFrame(_Frame):
         return self[list(cs)]
 
     def set_index(self, other, drop=True, sorted=False, npartitions=None,
-                  divisions=None, **kwargs):
+                  divisions=None, inplace=False, **kwargs):
         """Set the DataFrame index (row labels) using an existing column
 
         This realigns the dataset to be sorted by a new column.  This can have a
@@ -2655,6 +2655,9 @@ class DataFrame(_Frame):
             that if ``sorted=True``, specified divisions are assumed to match
             the existing partitions in the data. If this is untrue, you should
             leave divisions empty and call ``repartition`` after ``set_index``.
+        inplace : bool, optional
+            Modifying the DataFrame in place is not supported by Dask.
+            Defaults to False.
         compute: bool
             Whether or not to trigger an immediate computation. Defaults to False.
 
@@ -2673,6 +2676,8 @@ class DataFrame(_Frame):
         >>> divisions = pd.date_range('2000', '2010', freq='1D')
         >>> df2 = df.set_index('timestamp', sorted=True, divisions=divisions)  # doctest: +SKIP
         """
+        if inplace:
+            raise NotImplementedError("The inplace= keyword is not supported")
         pre_sorted = sorted
         del sorted
 
diff --git a/dask/dataframe/groupby.py b/dask/dataframe/groupby.py
index a0683cfc6..b115833b4 100644
--- a/dask/dataframe/groupby.py
+++ b/dask/dataframe/groupby.py
@@ -283,16 +283,21 @@ def _nunique_df_chunk(df, *index, **kwargs):
     name = kwargs.pop('name')
 
     g = _groupby_raise_unaligned(df, by=index)
-    grouped = g[[name]].apply(pd.DataFrame.drop_duplicates)
-
-    # we set the index here to force a possibly duplicate index
-    # for our reduce step
-    if isinstance(levels, list):
-        grouped.index = pd.MultiIndex.from_arrays([
-            grouped.index.get_level_values(level=level) for level in levels
-        ])
+    if len(df) > 0:
+        grouped = g[[name]].apply(pd.DataFrame.drop_duplicates)
+        # we set the index here to force a possibly duplicate index
+        # for our reduce step
+        if isinstance(levels, list):
+            grouped.index = pd.MultiIndex.from_arrays([
+                grouped.index.get_level_values(level=level) for level in levels
+            ])
+        else:
+            grouped.index = grouped.index.get_level_values(level=levels)
     else:
-        grouped.index = grouped.index.get_level_values(level=levels)
+        # Manually create empty version, since groupby-apply for empty frame
+        # results in df with no columns
+        grouped = g[[name]].nunique()
+        grouped = grouped.astype(df.dtypes[grouped.columns].to_dict())
 
     return grouped
 
diff --git a/dask/dataframe/utils.py b/dask/dataframe/utils.py
index ca00d6260..4321a1f50 100644
--- a/dask/dataframe/utils.py
+++ b/dask/dataframe/utils.py
@@ -398,13 +398,12 @@ def _nonempty_index(idx):
                                  name=idx.name)
     elif typ is pd.CategoricalIndex:
         if len(idx.categories) == 0:
-            data = _nonempty_index(idx.categories)
-            cats = None
+            data = pd.Categorical(_nonempty_index(idx.categories),
+                                  ordered=idx.ordered)
         else:
-            data = _nonempty_index(_nonempty_index(idx.categories))
-            cats = idx.categories
-        return pd.CategoricalIndex(data, categories=cats,
-                                   ordered=idx.ordered, name=idx.name)
+            data = pd.Categorical.from_codes(
+                [-1, 0], categories=idx.categories, ordered=idx.ordered)
+        return pd.CategoricalIndex(data, name=idx.name)
     elif typ is pd.MultiIndex:
         levels = [_nonempty_index(l) for l in idx.levels]
         labels = [[0, 0] for i in idx.levels]
diff --git a/dask/utils.py b/dask/utils.py
index ac58dfcb1..73aff6372 100644
--- a/dask/utils.py
+++ b/dask/utils.py
@@ -558,6 +558,25 @@ def funcname(func):
         return str(func)
 
 
+def typename(typ):
+    """
+    Return the name of a type
+
+    Examples
+    --------
+    >>> typename(int)
+    'int'
+
+    >>> from dask.core import literal
+    >>> typename(literal)
+    'dask.core.literal'
+    """
+    if not typ.__module__ or typ.__module__ == 'builtins':
+        return typ.__name__
+    else:
+        return typ.__module__ + '.' + typ.__name__
+
+
 def ensure_bytes(s):
     """ Turn string or bytes to bytes
 
diff --git a/docs/source/develop.rst b/docs/source/develop.rst
index 3e0ad25f7..d22755cb1 100644
--- a/docs/source/develop.rst
+++ b/docs/source/develop.rst
@@ -88,7 +88,7 @@ environment on your computer to compile them when installing with ``pip``::
 Install Dask and dependencies::
 
    cd dask
-   pip install -e .[complete]
+   pip install -e ".[complete]"
 
 For development, Dask uses the following additional dependencies::
 


Test patch:
diff --git a/dask/array/tests/test_masked.py b/dask/array/tests/test_masked.py
index 6608d2cc4..222e46c5c 100644
--- a/dask/array/tests/test_masked.py
+++ b/dask/array/tests/test_masked.py
@@ -348,3 +348,17 @@ def test_average_weights_with_masked_array():
     da_avg = da.ma.average(d_a, weights=d_weights, axis=1)
 
     assert_eq(np_avg, da_avg)
+
+
+def test_arithmetic_results_in_masked():
+    mask = np.array([[True, False],
+                     [True, True],
+                     [False, True]])
+    x = np.arange(6).reshape((3, 2))
+    masked = np.ma.array(x, mask=mask)
+    dx = da.from_array(x, chunks=(2, 2))
+
+    res = dx + masked
+    sol = x + masked
+    assert_eq(res, sol)
+    assert isinstance(res.compute(), np.ma.masked_array)
diff --git a/dask/bag/tests/test_bag.py b/dask/bag/tests/test_bag.py
index 116b41b99..c6a46be60 100644
--- a/dask/bag/tests/test_bag.py
+++ b/dask/bag/tests/test_bag.py
@@ -1314,3 +1314,11 @@ def test_bag_paths():
     assert b.to_textfiles('foo*') == ['foo0', 'foo1']
     os.remove('foo0')
     os.remove('foo1')
+
+
+def test_map_keynames():
+    b = db.from_sequence([1, 2, 3])
+    d = dict(b.map(inc).__dask_graph__())
+    assert 'inc' in map(dask.utils.key_split, d)
+
+    assert set(b.map(inc).__dask_graph__()) != set(b.map_partitions(inc).__dask_graph__())
diff --git a/dask/dataframe/tests/test_dataframe.py b/dask/dataframe/tests/test_dataframe.py
index 0476bf8a3..c712c2eb8 100644
--- a/dask/dataframe/tests/test_dataframe.py
+++ b/dask/dataframe/tests/test_dataframe.py
@@ -3380,3 +3380,12 @@ def test_has_parallel_type():
     assert has_parallel_type(pd.DataFrame())
     assert has_parallel_type(pd.Series())
     assert not has_parallel_type(123)
+
+
+def test_meta_error_message():
+    with pytest.raises(TypeError) as info:
+        dd.DataFrame({('x', 1): 123}, 'x', pd.Series(), [None, None])
+
+    assert 'Series' in str(info.value)
+    assert 'DataFrame' in str(info.value)
+    assert 'pandas' in str(info.value)
diff --git a/dask/dataframe/tests/test_groupby.py b/dask/dataframe/tests/test_groupby.py
index 83309b00d..075fcec68 100644
--- a/dask/dataframe/tests/test_groupby.py
+++ b/dask/dataframe/tests/test_groupby.py
@@ -405,10 +405,22 @@ def test_groupby_set_index():
                   lambda: ddf.groupby(df.index.month, as_index=False))
 
 
-def test_split_apply_combine_on_series():
-    pdf = pd.DataFrame({'a': [1, 2, 6, 4, 4, 6, 4, 3, 7],
-                        'b': [4, 2, 7, 3, 3, 1, 1, 1, 2]},
-                       index=[0, 1, 3, 5, 6, 8, 9, 9, 9])
+@pytest.mark.parametrize('empty', [
+    pytest.param(True, marks=pytest.mark.skipif(PANDAS_VERSION < '0.21.0',
+                 reason="Empty groupby-reductions fail for older pandas")),
+    pytest.param(False)
+])
+def test_split_apply_combine_on_series(empty):
+    if empty:
+        pdf = pd.DataFrame({'a': [1.], 'b': [1.]}, index=[0]).iloc[:0]
+        # There's a bug in pandas where df.groupby(...).var(ddof=0) results in
+        # no columns. Just skip these checks for now.
+        ddofs = []
+    else:
+        ddofs = [0, 1, 2]
+        pdf = pd.DataFrame({'a': [1, 2, 6, 4, 4, 6, 4, 3, 7],
+                            'b': [4, 2, 7, 3, 3, 1, 1, 1, 2]},
+                           index=[0, 1, 3, 5, 6, 8, 9, 9, 9])
     ddf = dd.from_pandas(pdf, npartitions=3)
 
     for ddkey, pdkey in [('b', 'b'), (ddf.b, pdf.b), (ddf.b + 1, pdf.b + 1)]:
@@ -420,7 +432,7 @@ def test_split_apply_combine_on_series():
         assert_eq(ddf.groupby(ddkey).a.size(), pdf.groupby(pdkey).a.size())
         assert_eq(ddf.groupby(ddkey).a.first(), pdf.groupby(pdkey).a.first())
         assert_eq(ddf.groupby(ddkey).a.last(), pdf.groupby(pdkey).a.last())
-        for ddof in [0, 1, 2]:
+        for ddof in ddofs:
             assert_eq(ddf.groupby(ddkey).a.var(ddof),
                       pdf.groupby(pdkey).a.var(ddof))
             assert_eq(ddf.groupby(ddkey).a.std(ddof),
@@ -435,7 +447,7 @@ def test_split_apply_combine_on_series():
         assert_eq(ddf.groupby(ddkey).first(), pdf.groupby(pdkey).first())
         assert_eq(ddf.groupby(ddkey).last(), pdf.groupby(pdkey).last())
 
-        for ddof in [0, 1, 2]:
+        for ddof in ddofs:
             assert_eq(ddf.groupby(ddkey).var(ddof),
                       pdf.groupby(pdkey).var(ddof), check_dtype=False)
             assert_eq(ddf.groupby(ddkey).std(ddof),
@@ -450,7 +462,7 @@ def test_split_apply_combine_on_series():
         assert_eq(ddf.a.groupby(ddkey).first(), pdf.a.groupby(pdkey).first(), check_names=False)
         assert_eq(ddf.a.groupby(ddkey).last(), pdf.a.groupby(pdkey).last(), check_names=False)
 
-        for ddof in [0, 1, 2]:
+        for ddof in ddofs:
             assert_eq(ddf.a.groupby(ddkey).var(ddof),
                       pdf.a.groupby(pdkey).var(ddof))
             assert_eq(ddf.a.groupby(ddkey).std(ddof),
@@ -495,7 +507,7 @@ def test_split_apply_combine_on_series():
         assert_eq(ddf.groupby(ddf.a > i).first(), pdf.groupby(pdf.a > i).first())
         assert_eq(ddf.groupby(ddf.a > i).last(), pdf.groupby(pdf.a > i).last())
 
-        for ddof in [0, 1, 2]:
+        for ddof in ddofs:
             assert_eq(ddf.groupby(ddf.b > i).std(ddof),
                       pdf.groupby(pdf.b > i).std(ddof))
 
@@ -520,7 +532,7 @@ def test_split_apply_combine_on_series():
         assert_eq(ddf.groupby(ddkey).first(), pdf.groupby(pdkey).first())
         assert_eq(ddf.groupby(ddkey).last(), pdf.groupby(pdkey).last())
 
-        for ddof in [0, 1, 2]:
+        for ddof in ddofs:
             assert_eq(ddf.groupby(ddkey).b.std(ddof),
                       pdf.groupby(pdkey).b.std(ddof))
 
diff --git a/dask/dataframe/tests/test_shuffle.py b/dask/dataframe/tests/test_shuffle.py
index 58cb823e8..9e9980412 100644
--- a/dask/dataframe/tests/test_shuffle.py
+++ b/dask/dataframe/tests/test_shuffle.py
@@ -782,3 +782,17 @@ def test_set_index_does_not_repeat_work_due_to_optimizations(npartitions):
     ddf.set_index('x', npartitions=npartitions)
     ntimes = next(count)
     assert ntimes == nparts
+
+
+def test_set_index_errors_with_inplace_kwarg():
+    df = pd.DataFrame({
+        'a': [9, 8, 7],
+        'b': [6, 5, 4],
+        'c': [3, 2, 1]
+    })
+    ddf = dd.from_pandas(df, npartitions=1)
+
+    ddf.set_index('a')
+
+    with pytest.raises(NotImplementedError):
+        ddf.set_index('a', inplace=True)
diff --git a/dask/dataframe/tests/test_utils_dataframe.py b/dask/dataframe/tests/test_utils_dataframe.py
index 603331329..081246014 100644
--- a/dask/dataframe/tests/test_utils_dataframe.py
+++ b/dask/dataframe/tests/test_utils_dataframe.py
@@ -209,7 +209,7 @@ def test_meta_nonempty_index():
     assert res.freq == idx.freq
     assert res.name == idx.name
 
-    idx = pd.CategoricalIndex(['a'], ['a', 'b'], ordered=True, name='foo')
+    idx = pd.CategoricalIndex(['xyx'], ['xyx', 'zzz'], ordered=True, name='foo')
     res = meta_nonempty(idx)
     assert type(res) is pd.CategoricalIndex
     assert (res.categories == idx.categories).all()
@@ -233,7 +233,7 @@ def test_meta_nonempty_index():
     assert res.names == idx.names
 
     levels = [pd.Int64Index([1], name='a'),
-              pd.CategoricalIndex(data=['b'], categories=['b'], name='b'),
+              pd.CategoricalIndex(data=['xyx'], categories=['xyx'], name='b'),
               pd.TimedeltaIndex([np.timedelta64(1, 'D')], name='timedelta')]
     idx = pd.MultiIndex(levels=levels, labels=[[0], [0], [0]], names=['a', 'b', 'timedelta'])
     res = meta_nonempty(idx)
diff --git a/dask/tests/test_base.py b/dask/tests/test_base.py
index 5d80253a2..82fb21e96 100644
--- a/dask/tests/test_base.py
+++ b/dask/tests/test_base.py
@@ -511,17 +511,17 @@ def test_compute_no_opt():
     keys = []
     with Callback(pretask=lambda key, *args: keys.append(key)):
         o.compute(scheduler='single-threaded', optimize_graph=False)
-    assert len([k for k in keys if '-mul-' in k[0]]) == 4
-    assert len([k for k in keys if '-add-' in k[0]]) == 4
+    assert len([k for k in keys if 'mul' in k[0]]) == 4
+    assert len([k for k in keys if 'add' in k[0]]) == 4
     # Check that without the kwarg, the optimization does happen
     keys = []
     with Callback(pretask=lambda key, *args: keys.append(key)):
         o.compute(scheduler='single-threaded')
     # Names of fused tasks have been merged, and the original key is an alias.
     # Otherwise, the lengths below would be 4 and 0.
-    assert len([k for k in keys if '-mul-' in k[0]]) == 8
-    assert len([k for k in keys if '-add-' in k[0]]) == 4
-    assert len([k for k in keys if 'add-map-mul' in k[0]]) == 4  # See? Renamed
+    assert len([k for k in keys if 'mul' in k[0]]) == 8
+    assert len([k for k in keys if 'add' in k[0]]) == 4
+    assert len([k for k in keys if 'add-from_sequence-mul' in k[0]]) == 4  # See? Renamed
 
 
 @pytest.mark.skipif('not da')