Repository: dask/dask

Issue:
DataFrame.groupby.mean fails with multiple categoricals
Seems closely related to issue #2510, but it looks like there's still a problem with `groupby.mean`.  Here's a quick example:

```
import dask.dataframe as dd

dt = {"col1": "category",
      "col2": "category",
      "col3": int}
df = dd.read_csv("data.csv", dtype=dt)

df.compute()
#   col1 col2  col3
# 0    a    y     0
# 1    a    z     1
# 2    b    y     0
# 3    b    z     1

# Gives an error
df.groupby(by=["col1", "col2"])["col3"].mean().compute()
# ValueError: Level values must be unique: [nan, nan] on level 0

# Taking a sum works
df.groupby(by=["col1", "col2"])["col3"].sum().compute()
# col1  col2
# a     y       0
#       z       1
# b     y       0
#       z       1
# Name: col3, dtype: int64

# Grouping by only one column also works
df.groupby(by="col1")["col3"].mean().compute()
# col1
# a    0.5
# b    0.5
# Name: col3, dtype: float64

# Use object dtype instead
df = dd.read_csv("data.csv")
df.dtypes
# col1    object
# col2    object
# col3     int64
# dtype: object

# Works now
df.groupby(by=["col1", "col2"]).mean().compute()
#            col3
# col1 col2      
# a    y      0.0
#      z      1.0
# b    y      0.0
#      z      1.0
```

This is `dask` 0.17.5 and `pandas` 0.24.0.

Discussion:

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/dataframe/core.py b/dask/dataframe/core.py
index 0245a4ca7..d1243ea68 100644
--- a/dask/dataframe/core.py
+++ b/dask/dataframe/core.py
@@ -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/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/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/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)