Repository: dask/dask

Issue:
NumPy's histogram's normed argument deprecated
Saw the following NumPy deprecation warning in [the CI build]( https://travis-ci.org/dask/dask/jobs/502248855#L1701-L1702 ) for the Dask Array tests. Looks like NumPy deprecated the `normed` argument of `histogram` in 1.15.0 with PR ( https://github.com/numpy/numpy/pull/11352 ). The changelog entry notes this was deprecated for much longer. Would be good to stop using the `normed` argument and also deprecate on our end as well.

```
dask/array/tests/test_routines.py::test_histogram_extra_args_and_shapes
  /Users/kirkhamj/Developer/PyCharmCE/dask/dask/array/tests/test_routines.py:590: VisibleDeprecationWarning: Passing `normed=True` on non-uniform bins has always been broken, and computes neither the probability density function nor the probability mass function. The result is only correct if the bins are uniform, when density=True will produce the same result anyway. The argument will be removed in a future version of numpy.
    np.histogram(v, bins=bins, normed=True)[0])
```

Discussion:

Patch:
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),