Repository: dask/dask

Issue:
Dataframe correlation function uses too much memory
Whenever I am calculating corr on a dask data frame with a larger but not enormous number of columns (200-2000) my worker memory is blowing up and getting  a MemoryError at one of two lines:

```
from /dask/dataframe/core.py

3885     x = np.where(keep, mat[:, None, :], np.nan)
3886     sums = np.nansum(x, 0)
```

This is reliably reproduced  with the code below, exact breaking point will obviously vary with worker,  I'm currently using 4 thread, 24GB workers.

```
arr = da.random.randint(100, size=(10_000, 500), chunks=1_000)
df  = dd.from_array(arr, columns=[f'col_{i}' for i in range(arr.shape[1])])
df.to_parquet(path)
df = dd.read_parquet(path)
corr = df.corr().compute()
```

Workers are reporting " distributed.worker - WARNING - Memory use is high but worker has no data to store to disk.  Perhaps some other process is leaking memory?  Process memory: 22.55 GB -- Worker memory limit: 24.00 GB"

Dask will break with memory errors at sizes small enough to process the entire calculation in memory with Pandas. I will need to scale to 10's of millions of rows and 1-3 thousand columns so staying in memory doesn't look to a be a good option.

Discussion:
mrocklin: Thanks @garaud .  I'm curious, do you have any information on how much memory use this takes up and on how much slower it is?  

Patch:
diff --git a/continuous_integration/travis/install.sh b/continuous_integration/travis/install.sh
index e01bd1d57..195db9efb 100644
--- a/continuous_integration/travis/install.sh
+++ b/continuous_integration/travis/install.sh
@@ -51,7 +51,7 @@ conda install -q -c conda-forge \
     partd \
     psutil \
     pytables \
-    "pytest<=3.1.1" \
+    "pytest<=3.10" \
     requests \
     scikit-image \
     scikit-learn \
diff --git a/dask/array/top.py b/dask/array/top.py
index 4f1cad48a..bbd601f5c 100644
--- a/dask/array/top.py
+++ b/dask/array/top.py
@@ -455,6 +455,16 @@ def atop(func, out_ind, *args, **kwargs):
     adjust_chunks = kwargs.pop('adjust_chunks', None)
     new_axes = kwargs.get('new_axes', {})
 
+    # Input Validation
+    if len(set(out_ind)) != len(out_ind):
+        raise ValueError("Repeated elements not allowed in output index",
+                         [k for k, v in toolz.frequencies(out_ind).items() if v > 1])
+    new = (set(out_ind)
+           - {a for arg in args[1::2] if arg is not None for a in arg}
+           - set(new_axes or ()))
+    if new:
+        raise ValueError("Unknown dimension", new)
+
     from .core import Array, unify_chunks, normalize_arg
 
     if dtype is None:
diff --git a/dask/dataframe/core.py b/dask/dataframe/core.py
index 6959539a3..1d615bd2b 100644
--- a/dask/dataframe/core.py
+++ b/dask/dataframe/core.py
@@ -3893,18 +3893,27 @@ def cov_corr(df, min_periods=None, corr=False, scalar=False, split_every=False):
 
 
 def cov_corr_chunk(df, corr=False):
-    """Chunk part of a covariance or correlation computation"""
-    mat = df.astype('float64', copy=False).values
-    mask = np.isfinite(mat)
-    keep = np.bitwise_and(mask[:, None, :], mask[:, :, None])
-
-    x = np.where(keep, mat[:, None, :], np.nan)
-    sums = np.nansum(x, 0)
-    counts = keep.astype('int').sum(0)
+    """Chunk part of a covariance or correlation computation
+    """
+    shape = (df.shape[1], df.shape[1])
+    sums = np.zeros(shape)
+    counts = np.zeros(shape)
+    df = df.astype('float64', copy=False)
+    for idx, col in enumerate(df):
+        mask = df[col].notnull()
+        sums[idx] = df[mask].sum().values
+        counts[idx] = df[mask].count().values
     cov = df.cov().values
     dtype = [('sum', sums.dtype), ('count', counts.dtype), ('cov', cov.dtype)]
     if corr:
-        m = np.nansum((x - sums / np.where(counts, counts, np.nan)) ** 2, 0)
+        mu = (sums / counts).T
+        m = np.zeros(shape)
+        mask = df.isnull().values
+        for idx, x in enumerate(df):
+            mu_discrepancy = np.subtract.outer(df[x], mu[idx]) ** 2
+            mu_discrepancy[mask] = np.nan
+            m[idx] = np.nansum(mu_discrepancy, axis=0)
+        m = m.T
         dtype.append(('m', m.dtype))
 
     out = np.empty(counts.shape, dtype=dtype)
diff --git a/dask/dot.py b/dask/dot.py
index e5fedfbd9..6235bdf21 100644
--- a/dask/dot.py
+++ b/dask/dot.py
@@ -6,7 +6,7 @@ from functools import partial
 
 from .compatibility import apply
 from .core import istask, get_dependencies, ishashable
-from .utils import funcname, import_required
+from .utils import funcname, import_required, key_split
 
 
 graphviz = import_required("graphviz", "Drawing dask graphs requires the "
@@ -104,6 +104,25 @@ def label(x, cache=None):
     return s
 
 
+def box_label(key):
+    """ Label boxes in graph by chunk index
+
+    >>> box_label(('x', 1, 2, 3))
+    '(1, 2, 3)'
+    >>> box_label(('x', 123))
+    '123'
+    >>> box_label('x')
+    ''
+    """
+    if isinstance(key, tuple):
+        key = key[1:]
+        if len(key) == 1:
+            [key] = key
+        return str(key)
+    else:
+        return ""
+
+
 def to_graphviz(dsk, data_attributes=None, function_attributes=None,
                 rankdir='BT', graph_attr={}, node_attr=None, edge_attr=None, **kwargs):
     if data_attributes is None:
@@ -119,14 +138,13 @@ def to_graphviz(dsk, data_attributes=None, function_attributes=None,
                          edge_attr=edge_attr)
 
     seen = set()
-    cache = {}
 
     for k, v in dsk.items():
         k_name = name(k)
         if k_name not in seen:
             seen.add(k_name)
             attrs = data_attributes.get(k, {})
-            attrs.setdefault('label', label(k, cache=cache))
+            attrs.setdefault('label', box_label(k))
             attrs.setdefault('shape', 'box')
             g.node(k_name, **attrs)
 
@@ -135,7 +153,7 @@ def to_graphviz(dsk, data_attributes=None, function_attributes=None,
             if func_name not in seen:
                 seen.add(func_name)
                 attrs = function_attributes.get(k, {})
-                attrs.setdefault('label', task_label(v))
+                attrs.setdefault('label', key_split(k))
                 attrs.setdefault('shape', 'circle')
                 g.node(func_name, **attrs)
             g.edge(func_name, k_name)
@@ -145,7 +163,7 @@ def to_graphviz(dsk, data_attributes=None, function_attributes=None,
                 if dep_name not in seen:
                     seen.add(dep_name)
                     attrs = data_attributes.get(dep, {})
-                    attrs.setdefault('label', label(dep, cache=cache))
+                    attrs.setdefault('label', box_label(dep))
                     attrs.setdefault('shape', 'box')
                     g.node(dep_name, **attrs)
                 g.edge(dep_name, func_name)
diff --git a/docs/source/array-design.rst b/docs/source/array-design.rst
index 0da3024cc..bc10c3390 100644
--- a/docs/source/array-design.rst
+++ b/docs/source/array-design.rst
@@ -10,15 +10,15 @@ Overview
    :alt: A dask array
 
 Dask arrays define a large array with a grid of blocks of smaller arrays.
-These arrays may be concrete, or functions that produce arrays.  We define a
-Dask array with the following components
+These arrays may be actual arrays or functions that produce arrays. We 
+define a Dask array with the following components:
 
 *  A Dask graph with a special set of keys designating blocks
    such as ``('x', 0, 0), ('x', 0, 1), ...`` (See :doc:`Dask graph
-   documentation <graphs>` for more details.)
+   documentation <graphs>` for more details)
 *  A sequence of chunk sizes along each dimension called ``chunks``,
    for example ``((5, 5, 5, 5), (8, 8, 8))``
-*  A name to identify which keys in the dask graph refer to this array, like
+*  A name to identify which keys in the Dask graph refer to this array, like
    ``'x'``
 *  A NumPy dtype
 
@@ -48,11 +48,11 @@ Example
 Keys of the Dask graph
 ----------------------
 
-By special convention we refer to each block of the array with a tuple of the
-form ``(name, i, j, k)`` for ``i, j, k`` being the indices of the block,
-ranging from ``0`` to the number of blocks in that dimension.  The dask graph
-must hold key-value pairs referring to these keys.  It likely also holds other
-key-value pairs required to eventually compute the desired values, for example
+By special convention, we refer to each block of the array with a tuple of the
+form ``(name, i, j, k)``, with ``i, j, k`` being the indices of the block
+ranging from ``0`` to the number of blocks in that dimension.  The Dask graph
+must hold key-value pairs referring to these keys.  Moreover, it likely also 
+holds other key-value pairs required to eventually compute the desired values:
 
 .. code-block:: python
 
@@ -66,31 +66,32 @@ key-value pairs required to eventually compute the desired values, for example
    }
 
 The name of an ``Array`` object can be found in the ``name`` attribute.  One
-can get a nested list of keys with the ``.__dask_keys__()`` method.  One can
-flatten down this list with ``dask.array.core.flatten()``; this is sometimes
+can get a nested list of keys with the ``.__dask_keys__()`` method.  Additionally, 
+one can flatten down this list with ``dask.array.core.flatten()``. This is sometimes
 useful when building new dictionaries.
 
 Chunks
 ------
 
-We also store the size of each block along each axis.  This is a tuple of
-tuples such that the length of the outer tuple is equal to the dimension and
-the lengths of the inner tuples are equal to the number of blocks along each
-dimension.  In the example illustrated above this value is as follows::
+We also store the size of each block along each axis.  This is composed of 
+a tuple of tuples such that the length of the outer tuple is equal to the 
+number of dimensions of the array, and the lengths of the inner tuples are 
+equal to the number of blocks along each dimension.  In the example illustrated 
+above this value is as follows::
 
     chunks = ((5, 5, 5, 5), (8, 8, 8))
 
 Note that these numbers do not necessarily need to be regular.  We often create
 regularly sized grids but blocks change shape after complex slicing.  Beware
 that some operations do expect certain symmetries in the block-shapes.  For
-example matrix multiplication requires that blocks on each side have
+example, matrix multiplication requires that blocks on each side have
 anti-symmetric shapes.
 
-Some ways in which ``chunks`` reflects properties of our array
+Some ways in which ``chunks`` reflects properties of our array:
 
-1.  ``len(x.chunks) == x.ndim``: The length of chunks is the number of dimensions
-2.  ``tuple(map(sum, x.chunks)) == x.shape``: The sum of each internal chunk, is the
-    length of that dimension.
+1.  ``len(x.chunks) == x.ndim``: the length of chunks is the number of dimensions
+2.  ``tuple(map(sum, x.chunks)) == x.shape``: the sum of each internal chunk is the
+    length of that dimension
 3.  The length of each internal chunk is the number of keys in that dimension.
     For instance, for ``chunks == ((a, b), (d, e, f))`` and name == ``'x'``
     our array has tasks with the following keys::
@@ -102,12 +103,12 @@ Some ways in which ``chunks`` reflects properties of our array
 Create an Array Object
 ----------------------
 
-So to create an ``da.Array`` object we need a dictionary with these special
-keys ::
+In order to create an ``da.Array`` object we need a dictionary with these special
+keys::
 
     dsk = {('x', 0, 0): ...}
 
-a name specifying to which keys this array refers ::
+a name specifying which keys this array refers to::
 
     name = 'x'
 
@@ -115,19 +116,19 @@ and a chunks tuple::
 
     chunks = ((5, 5, 5, 5), (8, 8, 8))
 
-Then one can construct an array::
+Then, using these elements, one can construct an array::
 
     x = da.Array(dsk, name, chunks)
 
-So ``dask.array`` operations update dask graphs, update dtypes, and track chunk
+In short, ``dask.array`` operations update Dask graphs, update dtypes, and track chunk
 shapes.
 
 
 Example - ``eye`` function
 --------------------------
 
-As an example lets build the ``np.eye`` function for ``dask.array`` to make the
-identity matrix
+As an example, lets build the ``np.eye`` function for ``dask.array`` to make the
+identity matrix:
 
 .. code-block:: python
 
diff --git a/docs/source/array-linear-operator.rst b/docs/source/array-linear-operator.rst
index 5e7a58737..6e78a0a56 100644
--- a/docs/source/array-linear-operator.rst
+++ b/docs/source/array-linear-operator.rst
@@ -1,7 +1,7 @@
 LinearOperator
 ==============
 
-Dask arrays implement the SciPy LinearOperator_ interface and so can be used
+Dask Array implements the SciPy LinearOperator_ interface and it can be used
 with any SciPy algorithm depending on that interface.
 
 Example
diff --git a/docs/source/array-slicing.rst b/docs/source/array-slicing.rst
index 4f8a34e80..b3a24896e 100644
--- a/docs/source/array-slicing.rst
+++ b/docs/source/array-slicing.rst
@@ -1,22 +1,22 @@
 Slicing
 =======
 
-Dask array supports most of the NumPy slicing syntax.  In particular it
+Dask Array supports most of the NumPy slicing syntax.  In particular, it
 supports the following:
 
-*  Slicing by integers and slices ``x[0, :5]``
-*  Slicing by lists/arrays of integers  ``x[[1, 2, 4]]``
-*  Slicing by lists/arrays of booleans ``x[[False, True, True, False, True]]``
-*  Slicing one :class:`~dask.array.Array` with a :class:`~dask.array.Array` of bools ``x[x > 0]``
+*  Slicing by integers and slices: ``x[0, :5]``
+*  Slicing by lists/arrays of integers: ``x[[1, 2, 4]]``
+*  Slicing by lists/arrays of booleans: ``x[[False, True, True, False, True]]``
+*  Slicing one :class:`~dask.array.Array` with an :class:`~dask.array.Array` of bools: ``x[x > 0]``
 *  Slicing one :class:`~dask.array.Array` with a zero or one-dimensional :class:`~dask.array.Array`
-   of ints ``a[b.argtopk(5)]``
+   of ints: ``a[b.argtopk(5)]``
 
-It does not currently support the following:
+However, it does not currently support the following:
 
-*  Slicing with lists in multiple axes  ``x[[1, 2, 3], [3, 2, 1]]``
+*  Slicing with lists in multiple axes: ``x[[1, 2, 3], [3, 2, 1]]``
 
    This is straightforward to add though.  If you have a use case then raise an
-   issue. Also users interested in this should take a look at
+   issue. Also, users interested in this should take a look at
    :attr:`~dask.array.Array.vindex`.
 
 *  Slicing one :class:`~dask.array.Array` with a multi-dimensional :class:`~dask.array.Array` of ints
@@ -24,26 +24,26 @@ It does not currently support the following:
 Efficiency
 ----------
 
-The normal dask schedulers are smart enough to compute only those blocks that
-are necessary to achieve the desired slicing.  So large operations may be cheap
+The normal Dask schedulers are smart enough to compute only those blocks that
+are necessary to achieve the desired slicing.  Hence, large operations may be cheap
 if only a small output is desired.
 
-In the example below we create a trillion element Dask array in million element
-blocks.  We then operate on the entire array and finally slice out only a
-portion of the output.
+In the example below, we create a Dask array with a trillion elements with million 
+element sized blocks.  We then operate on the entire array and finally slice out 
+only a portion of the output:
 
 .. code-block:: python
 
-   >>> Trillion element array of ones, in 1000 by 1000 blocks
+   >>> # Trillion element array of ones, in 1000 by 1000 blocks
    >>> x = da.ones((1000000, 1000000), chunks=(1000, 1000))
 
    >>> da.exp(x)[:1500, :1500]
    ...
 
 This only needs to compute the top-left four blocks to achieve the result.  We
-are still slightly wasteful on those blocks where we need only partial results.
-We are also a bit wasteful in that we still need to manipulate the dask-graph
+are slightly wasteful on those blocks where we need only partial results.  Moreover, 
+we are also a bit wasteful in that we still need to manipulate the Dask graph
 with a million or so tasks in it.  This can cause an interactive overhead of a
-second or two.
+second or two. 
 
 But generally, slicing works well.
diff --git a/docs/source/array-stack.rst b/docs/source/array-stack.rst
index 503c3d681..43d8422d3 100644
--- a/docs/source/array-stack.rst
+++ b/docs/source/array-stack.rst
@@ -6,7 +6,7 @@ think of as one large array.  This is common with geospatial data in which we
 might have many HDF5/NetCDF files on disk, one for every day, but we want to do
 operations that span multiple days.
 
-To solve this problem we use the functions ``da.stack``, ``da.concatenate``,
+To solve this problem, we use the functions ``da.stack``, ``da.concatenate``,
 and ``da.block``.
 
 Stack
@@ -63,7 +63,7 @@ Block
 -----
 
 We can handle a larger variety of cases with ``da.block`` as it allows
-concatenation to be applied over multiple dimensions at once. This is useful if
+concatenation to be applied over multiple dimensions at once.  This is useful if
 your chunks tile a space, for example if small squares tile a larger 2-D plane.
 
 .. code-block:: python
diff --git a/docs/source/array-stats.rst b/docs/source/array-stats.rst
index 53e8f1be7..c28a7f479 100644
--- a/docs/source/array-stats.rst
+++ b/docs/source/array-stats.rst
@@ -6,7 +6,7 @@ Dask Array implements a subset of the `scipy.stats`_ package.
 Statistical Functions
 ---------------------
 
-You can calculate various measures of an array including skewnes, kurtosis, and arbitrary moments.
+You can calculate various measures of an array including skewness, kurtosis, and arbitrary moments.
 
 .. code-block:: python
 
@@ -20,7 +20,7 @@ You can calculate various measures of an array including skewnes, kurtosis, and
 Statistical Tests
 -----------------
 
-You can perform basic statistical tests on dask arrays.
+You can perform basic statistical tests on Dask arrays.
 Each of these tests return a ``dask.delayed`` wrapping one of the scipy ``namedtuple``
 results.
 
diff --git a/docs/source/changelog.rst b/docs/source/changelog.rst
index 56d4cebc6..268ef8944 100644
--- a/docs/source/changelog.rst
+++ b/docs/source/changelog.rst
@@ -1,6 +1,45 @@
 Changelog
 =========
 
+0.20.1 / 2018-11-09
+-------------------
+
+Array
++++++
+
+-  Only allocate the result space in wrapped_pad_func (:pr:`4153`) `John A Kirkham`_
+-  Generalize expand_pad_width to expand_pad_value (:pr:`4150`) `John A Kirkham`_
+-  Test da.pad with 2D linear_ramp case (:pr:`4162`) `John A Kirkham`_
+-  Fix import for broadcast_to. (:pr:`4168`) `samc0de`_
+-  Rewrite Dask Array's `pad` to add only new chunks (:pr:`4152`) `John A Kirkham`_
+-  Validate index inputs to atop (:pr:`4182`) `Matthew Rocklin`_
+
+Core
+++++
+
+-  Dask.config set and get normalize underscores and hyphens (:pr:`4143`) `James Bourbeau`_
+-  Only subs on core collections, not subclasses (:pr:`4159`) `Matthew Rocklin`_
+-  Add block_size=0 option to HTTPFileSystem. (:pr:`4171`) `Martin Durant`_
+-  Add traverse support for dataclasses (:pr:`4165`) `Armin Berres`_
+-  Avoid optimization on sharedicts without dependencies (:pr:`4181`) `Matthew Rocklin`_
+-  Update the pytest version for TravisCI (:pr:`4189`) `Damien Garaud`_
+-  Use key_split rather than funcname in visualize names (:pr:`4160`) `Matthew Rocklin`_
+
+Dataframe
++++++++++
+
+-  Add fix for  DataFrame.__setitem__ for index (:pr:`4151`) `Anderson Banihirwe`_
+-  Fix column choice when passing list of files to fastparquet (:pr:`4174`) `Martin Durant`_
+-  Pass engine_kwargs from read_sql_table to sqlalchemy (:pr:`4187`) `Damien Garaud`_
+
+Documentation
++++++++++++++
+
+-  Fix documentation in Delayed best practices example that returned an empty list (:pr:`4147`) `Jonathan Fraine`_
+-  Copy edit documentation (:pr:`4164`) (:pr:`4175`) (:pr:`4185`) (:pr:`4192`) (:pr:`4191`) (:pr:`4190`) (:pr:`4180`) `Miguel Farrajota`_
+-  Fix typo in docstring (:pr:`4183`) `Carlos Valiente`_
+
+
 0.20.0 / 2018-10-26
 -------------------
 
@@ -1553,3 +1592,8 @@ Other
 .. _`Justin Dennison`: https://github.com/justin1dennison
 .. _`Antonino Ingargiola`: https://github.com/tritemio
 .. _`TakaakiFuruse`: https://github.com/TakaakiFuruse
+.. _`samc0de`: https://github.com/samc0de
+.. _`Armin Berres`: https://github.com/aberres
+.. _`Damien Garaud`: https://github.com/geraud
+.. _`Jonathan Fraine`: https://github.com/exowanderer
+.. _`Carlos Valiente`: https://github.com/carletes
diff --git a/docs/source/dataframe.rst b/docs/source/dataframe.rst
index be03b0030..84680866d 100644
--- a/docs/source/dataframe.rst
+++ b/docs/source/dataframe.rst
@@ -1,4 +1,4 @@
-Dataframe
+DataFrame
 =========
 
 .. toctree::
@@ -13,17 +13,17 @@ Dataframe
    dataframe-indexing.rst
    dataframe-extend.rst
 
-A Dask DataFrame is a large parallel dataframe composed of many smaller Pandas
-dataframes, split along the index.  These pandas dataframes may live on disk
+A Dask DataFrame is a large parallel DataFrame composed of many smaller Pandas
+DataFrames, split along the index.  These Pandas DataFrames may live on disk
 for larger-than-memory computing on a single machine, or on many different
-machines in a cluster.  One Dask dataframe operation triggers many operations
-on the constituent Pandas dataframes.
+machines in a cluster.  One Dask DataFrame operation triggers many operations
+on the constituent Pandas DataFrames.
 
 Design
 ------
 
-Dask dataframes coordinate many Pandas DataFrames/Series arranged along the
-index.  Dask.dataframe is partitioned *row-wise*, grouping rows by index value
+Dask DataFrames coordinate many Pandas DataFrames/Series arranged along the
+index.  A Dask DataFrame is partitioned *row-wise*, grouping rows by index value
 for efficiency.  These Pandas objects may live on disk or on other machines.
 
 .. image:: images/dask-dataframe.svg
@@ -31,12 +31,12 @@ for efficiency.  These Pandas objects may live on disk or on other machines.
    :width: 40%
 
 
-Dask.dataframe copies the Pandas API
+Dask DataFrame copies the Pandas API
 ------------------------------------
 
 Because the ``dask.dataframe`` application programming interface (API) is a
-subset of the Pandas API it should be familiar to Pandas users.  There are some
-slight alterations due to the parallel nature of dask:
+subset of the Pandas API, it should be familiar to Pandas users.  There are some
+slight alterations due to the parallel nature of Dask:
 
 .. code-block:: python
 
@@ -53,7 +53,7 @@ slight alterations due to the parallel nature of dask:
 
    >>> df2 = df[df.y == 'a'].x + 1
 
-As with all dask collections one triggers computation by calling the
+As with all Dask collections, one triggers computation by calling the
 ``.compute()`` method:
 
 .. code-block:: python
@@ -67,24 +67,24 @@ As with all dask collections one triggers computation by calling the
 Common Uses and Anti-Uses
 -------------------------
 
-Dask.dataframe is used in situations where Pandas is commonly needed, but when
-Pandas fails due to data size or computation speed.
+Dask DataFrame is used in situations where Pandas is commonly needed, usually when
+Pandas fails due to data size or computation speed:
 
 -  Manipulating large datasets, even when those datasets don't fit in memory
 -  Accelerating long computations by using many cores
 -  Distributed computing on large datasets with standard Pandas operations like
    groupby, join, and time series computations
 
-Dask dataframe may not be the best choice in the following situations:
+Dask DataFrame may not be the best choice in the following situations:
 
-*  If your dataset fits comfortably into RAM on your laptop then you may be
-   better off just using Pandas .  There may be simpler ways to improve
-   performance than through parallelism.
-*  If your dataset doesn't fit neatly into the Pandas tabular model then you
+*  If your dataset fits comfortably into RAM on your laptop, then you may be
+   better off just using Pandas.  There may be simpler ways to improve
+   performance than through parallelism
+*  If your dataset doesn't fit neatly into the Pandas tabular model, then you
    might find more use in :doc:`dask.bag <bag>` or :doc:`dask.array <array>`
-*  If you need functions that are not implemented in Dask dataframe then you
+*  If you need functions that are not implemented in Dask DataFrame, then you
    might want to look at :doc:`dask.delayed <delayed>` which offers more
-   flexibility.
+   flexibility
 *  If you need a proper database with all that databases offer you might prefer
    something like Postgres_
 
@@ -95,16 +95,16 @@ Dask dataframe may not be the best choice in the following situations:
 Scope
 -----
 
-Dask.dataframe covers a well-used portion of the Pandas API.
+Dask DataFrame covers a well-used portion of the Pandas API.
 The following class of computations works well:
 
 * Trivially parallelizable operations (fast):
-    *  Elementwise operations:  ``df.x + df.y``, ``df * df``
+    *  Element-wise operations:  ``df.x + df.y``, ``df * df``
     *  Row-wise selections:  ``df[df.x > 0]``
     *  Loc:  ``df.loc[4.0:10.5]``
     *  Common aggregations:  ``df.x.max()``, ``df.max()``
     *  Is in:  ``df[df.x.isin([1, 2, 3])]``
-    *  Datetime/string accessors:  ``df.timestamp.month``
+    *  Date time/string accessors:  ``df.timestamp.month``
 * Cleverly parallelizable operations (fast):
     *  groupby-aggregate (with common aggregations): ``df.groupby(df.x).y.max()``,
        ``df.groupby('x').max()``
@@ -116,26 +116,26 @@ The following class of computations works well:
        or ``dd.merge(df1, df2, on=['idx', 'x'])`` where ``idx`` is the index
        name for both ``df1`` and ``df2``
     *  Join with Pandas DataFrames: ``dd.merge(df1, df2, on='id')``
-    *  Elementwise operations with different partitions / divisions: ``df1.x + df2.y``
-    *  Datetime resampling: ``df.resample(...)``
+    *  Element-wise operations with different partitions / divisions: ``df1.x + df2.y``
+    *  Date time resampling: ``df.resample(...)``
     *  Rolling averages:  ``df.rolling(...)``
-    *  Pearson Correlations: ``df[['col1', 'col2']].corr()``
+    *  Pearson's correlation: ``df[['col1', 'col2']].corr()``
 * Operations requiring a shuffle (slow-ish, unless on index)
     *  Set index:  ``df.set_index(df.x)``
     *  groupby-apply not on index (with anything):  ``df.groupby(df.x).apply(myfunc)``
     *  Join not on the index:  ``dd.merge(df1, df2, on='name')``
 
-However Dask dataframe does not implement the entire Pandas interface.  Users
-expecting this will be disappointed.  Notably, Dask dataframe has the following
+However, Dask DataFrame does not implement the entire Pandas interface.  Users
+expecting this will be disappointed.  Notably, Dask DataFrame has the following
 limitations:
 
 1.  Setting a new index from an unsorted column is expensive
-2.  Many operations, like groupby-apply and join on unsorted columns require
+2.  Many operations like groupby-apply and join on unsorted columns require
     setting the index, which as mentioned above, is expensive
-3.  The Pandas API is very large.  Dask dataframe does not attempt to implement
+3.  The Pandas API is very large.  Dask DataFrame does not attempt to implement
     many Pandas features or any of the more exotic data structures like NDFrames
 4.  Operations that were slow on Pandas, like iterating through row-by-row,
-    remain slow on Dask dataframe
+    remain slow on Dask DataFrame
 
 See :doc:`DataFrame API documentation<dataframe-api>` for a more extensive list.
 
@@ -143,13 +143,13 @@ See :doc:`DataFrame API documentation<dataframe-api>` for a more extensive list.
 Execution
 ---------
 
-By default ``dask.dataframe`` uses the multi-threaded scheduler.
-This exposes some parallelism when Pandas or the underlying numpy operations
-release the global interpreter lock (GIL).  Generally Pandas is more GIL
+By default, Dask DataFrame uses the multi-threaded scheduler.
+This exposes some parallelism when Pandas or the underlying NumPy operations
+release the global interpreter lock (GIL).  Generally, Pandas is more GIL
 bound than NumPy, so multi-core speed-ups are not as pronounced for
-``dask.dataframe`` as they are for ``dask.array``.  This is changing, and
+Dask DataFrame as they are for Dask Array.  This is changing, and
 the Pandas development team is actively working on releasing the GIL.
 
-When dealing with text data you may see speedups by switching to the newer
+When dealing with text data, you may see speedups by switching to the newer
 :doc:`distributed scheduler <setup/single-distributed>` either on a cluster or
 single machine.
diff --git a/setup.cfg b/setup.cfg
index 5fc16180d..293bac751 100644
--- a/setup.cfg
+++ b/setup.cfg
@@ -13,6 +13,7 @@ ignore =
     E721,  # Comparing types instead of isinstance
     E731,  # Assigning lambda expression
     E741,  # Ambiguous variable names
+    W503,  # line break before binary operator
     W504,  # line break after binary operator
     F811,  # redefinition of unused 'loop' from line 10
 max-line-length = 120


Test patch:
diff --git a/dask/array/tests/test_atop.py b/dask/array/tests/test_atop.py
index 117a15b9c..db243f08b 100644
--- a/dask/array/tests/test_atop.py
+++ b/dask/array/tests/test_atop.py
@@ -325,15 +325,6 @@ def test_atop_chunks():
     assert_eq(y, np.ones((20, 10)))
 
 
-def test_atop_raises_on_incorrect_indices():
-    x = da.arange(5, chunks=3)
-    with pytest.raises(ValueError) as info:
-        da.atop(lambda x: x, 'ii', x, 'ii', dtype=int)
-
-    assert 'ii' in str(info.value)
-    assert '1' in str(info.value)
-
-
 def test_atop_numpy_arg():
     x = da.arange(10, chunks=(5,))
     y = np.arange(1000)
@@ -382,6 +373,23 @@ def test_namedtuple(tup):
     assert_eq(A, B)
 
 
+def test_validate_top_inputs():
+    A = da.random.random((20, 20), chunks=(10, 10))
+
+    with pytest.raises(ValueError) as info:
+        da.atop(inc, 'jk', A, 'ij', dtype=A.dtype)
+
+    assert 'unknown dimension' in str(info.value).lower()
+    assert 'k' in str(info.value)
+    assert 'j' not in str(info.value)
+
+    with pytest.raises(ValueError) as info:
+        da.atop(inc, 'ii', A, 'ij', dtype=A.dtype)
+
+    assert 'repeated' in str(info.value).lower()
+    assert 'i' in str(info.value)
+
+
 def test_gh_4176():
     def foo(A):
         return A[None, ...]
diff --git a/dask/tests/test_dot.py b/dask/tests/test_dot.py
index d59ab6ebf..a8d870895 100644
--- a/dask/tests/test_dot.py
+++ b/dask/tests/test_dot.py
@@ -76,9 +76,7 @@ def test_to_graphviz():
     g = to_graphviz(dsk)
     labels = list(filter(None, map(get_label, g.body)))
     assert len(labels) == 10        # 10 nodes total
-    funcs = set(('add', 'sum', 'neg'))
-    assert set(labels).difference(dsk) == funcs
-    assert set(labels).difference(funcs) == set(dsk)
+    assert set(labels) == {'c', 'd', 'e', 'f', '""'}
     shapes = list(filter(None, map(get_shape, g.body)))
     assert set(shapes) == set(('box', 'circle'))
 
@@ -89,10 +87,8 @@ def test_to_graphviz_custom():
         data_attributes={'a': {'shape': 'square'}},
         function_attributes={'c': {'label': 'neg_c', 'shape': 'ellipse'}},
     )
-    labels = list(filter(None, map(get_label, g.body)))
-    funcs = set(('add', 'sum', 'neg', 'neg_c'))
-    assert set(labels).difference(dsk) == funcs
-    assert set(labels).difference(funcs) == set(dsk)
+    labels = set(filter(None, map(get_label, g.body)))
+    assert labels == {'neg_c', 'd', 'e', 'f', '""'}
     shapes = list(filter(None, map(get_shape, g.body)))
     assert set(shapes) == set(('box', 'circle', 'square', 'ellipse'))