Repository: dask/dask

Issue:
Xarray handling regression in 1.1.2
Since dask 1.1.2, this crashes:

```python
import xarray as xr                                                                             
import dask.array as da                                                                         
da.mean(xr.DataArray([1, 2, 3.0])).compute()
```
The error raised:
`TypeError: nansum() got an unexpected keyword argument 'keepdims'`

I looks like a regression to me, but maybe I'm missing something ?

Works fine in 1.1.1

Relevant packages:
dask 1.1.2
xarray 0.11.3
numpy 1.16.2

Discussion:

Patch:
diff --git a/dask/array/core.py b/dask/array/core.py
index 7086c51e0..bdcf53956 100644
--- a/dask/array/core.py
+++ b/dask/array/core.py
@@ -3087,6 +3087,8 @@ def asanyarray(a):
         return a
     elif hasattr(a, 'to_dask_array'):
         return a.to_dask_array()
+    elif hasattr(a, 'data') and type(a).__module__.startswith('xarray.'):
+        return asanyarray(a.data)
     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):
diff --git a/dask/array/routines.py b/dask/array/routines.py
index 261f7501e..815de0747 100644
--- a/dask/array/routines.py
+++ b/dask/array/routines.py
@@ -609,6 +609,13 @@ def histogram(a, bins=None, range=None, normed=False, weights=None, density=None
         raise ValueError('Input array and weights must have the same '
                          'chunked structure')
 
+    if normed is not False:
+        raise ValueError(
+            "The normed= keyword argument has been deprecated. "
+            "Please use density instead. "
+            "See the numpy.histogram docstring for more information."
+        )
+
     if not np.iterable(bins):
         bin_token = bins
         mn, mx = range
@@ -619,7 +626,7 @@ def histogram(a, bins=None, range=None, normed=False, weights=None, density=None
         bins = np.linspace(mn, mx, bins + 1, endpoint=True)
     else:
         bin_token = bins
-    token = tokenize(a, bin_token, range, normed, weights, density)
+    token = tokenize(a, bin_token, range, weights, density)
 
     nchunks = len(list(flatten(a.__dask_keys__())))
     chunks = ((1,) * nchunks, (len(bins) - 1,))
@@ -654,12 +661,7 @@ def histogram(a, bins=None, range=None, normed=False, weights=None, density=None
         else:
             return n, bins
     else:
-        # deprecated, will be removed from Numpy 2.0
-        if normed:
-            db = from_array(np.diff(bins).astype(float), chunks=n.chunks)
-            return n / (n * db).sum(), bins
-        else:
-            return n, bins
+        return n, bins
 
 
 @wraps(np.cov)
diff --git a/dask/dataframe/core.py b/dask/dataframe/core.py
index 65b3873d4..4166acfc3 100644
--- a/dask/dataframe/core.py
+++ b/dask/dataframe/core.py
@@ -495,7 +495,8 @@ Dask Name: {name}, {task} tasks""".format(klass=self.__class__.__name__,
             Arguments and keywords to pass to the function. The partition will
             be the first argument, and these will be passed *after*. Arguments
             and keywords may contain ``Scalar``, ``Delayed`` or regular
-            python objects.
+            python objects. DataFrame-like args (both dask and pandas) will be
+            repartitioned to align (if necessary) before applying the function.
         $META
 
         Examples
@@ -3775,7 +3776,9 @@ def map_partitions(func, *args, **kwargs):
     args, kwargs :
         Arguments and keywords to pass to the function.  At least one of the
         args should be a Dask.dataframe. Arguments and keywords may contain
-        ``Scalar``, ``Delayed`` or regular python objects.
+        ``Scalar``, ``Delayed`` or regular python objects. DataFrame-like args
+        (both dask and pandas) will be repartitioned to align (if necessary)
+        before applying the function.
     $META
     """
     meta = kwargs.pop('meta', no_default)
diff --git a/docs/source/optimize.rst b/docs/source/optimize.rst
index 17a303a34..9a5604569 100644
--- a/docs/source/optimize.rst
+++ b/docs/source/optimize.rst
@@ -67,14 +67,18 @@ Here we are counting the occurrence of the words ``'orange``, ``'apple'``, and
 ``'pear'`` in the list of words, formatting an output string reporting the
 results, printing the output, and then returning the output string.
 
-To perform the computation, we pass the Dask graph and the desired output keys
-to a scheduler ``get`` function:
+To perform the computation, we first remove unnecessary components from the
+graph using the ``cull`` function and then pass the Dask graph and the desired
+output keys to a scheduler ``get`` function:
 
 .. code-block:: python
 
     >>> from dask.threaded import get
+    >>> from dask.optimization import cull
 
     >>> outputs = ['print1', 'print2']
+    >>> dsk2, _ = cull(dsk, outputs)  # remove unnecessary tasks from the graph
+
     >>> results = get(dsk, outputs)
     word list has 2 occurrences of apple, out of 7 words
     word list has 2 occurrences of orange, out of 7 words
@@ -84,10 +88,12 @@ to a scheduler ``get`` function:
      'word list has 2 occurrences of apple, out of 7 words')
 
 As can be seen above, the scheduler computed only the requested outputs
-(``'print3'`` was never computed). This is because the scheduler internally
-calls ``cull``, which removes the unnecessary tasks from the graph. Even though
-this is done internally in the scheduler, it can be beneficial to call it at
-the start of a series of optimizations to reduce the amount of work done in
+(``'print3'`` was never computed). This is because we called the
+``dask.optimization.cull`` function, which removes the unnecessary tasks from
+the graph.
+
+Culling is part of the default optimization pass of almost all collections.
+Often you want to call it somewhat early to reduce the amount of work done in
 later steps:
 
 .. code-block:: python


Test patch:
diff --git a/dask/array/tests/test_routines.py b/dask/array/tests/test_routines.py
index b14ef6864..558414c61 100644
--- a/dask/array/tests/test_routines.py
+++ b/dask/array/tests/test_routines.py
@@ -570,10 +570,6 @@ def test_histogram_extra_args_and_shapes():
 
     for v, bins, w in data:
         # density
-        assert_eq(da.histogram(v, bins=bins, normed=True)[0],
-                  np.histogram(v, bins=bins, normed=True)[0])
-
-        # normed
         assert_eq(da.histogram(v, bins=bins, density=True)[0],
                   np.histogram(v, bins=bins, density=True)[0])
 
@@ -585,6 +581,15 @@ def test_histogram_extra_args_and_shapes():
                   da.histogram(v, bins=bins, weights=w, density=True)[0])
 
 
+def test_histogram_normed_deprecation():
+    x = da.arange(10)
+    with pytest.raises(ValueError) as info:
+        da.histogram(x, bins=[1, 2, 3], normed=True)
+
+    assert 'density' in str(info.value)
+    assert 'deprecated' in str(info.value).lower()
+
+
 @pytest.mark.parametrize("bins, hist_range", [
     (None, None),
     (10, None),
diff --git a/dask/array/tests/test_xarray.py b/dask/array/tests/test_xarray.py
new file mode 100644
index 000000000..43ea55b7f
--- /dev/null
+++ b/dask/array/tests/test_xarray.py
@@ -0,0 +1,12 @@
+import pytest
+
+import dask.array as da
+from ..utils import assert_eq
+
+xr = pytest.importorskip('xarray')
+
+
+def test_xarray():
+    y = da.mean(xr.DataArray([1, 2, 3.0]))
+
+    assert_eq(y, y)