Repository: dask/dask

Issue:
`dask.array.nanmean()` generates unstable name (hash)
#### Code Sample

```python
import dask.array as da
import numpy as np

x = da.ones((5, 5), chunks=(2, 2))
x = da.nanmean(x, axis=0)
print(x.name)
```

#### Problem Description
Running the above sample three times outputs:
```
mean_agg-aggregate-8157871966f12f849b40951c495fef6f
mean_agg-aggregate-c6939e3b5eb3290f17252b895e10c7c3
mean_agg-aggregate-17e2ac663f1cdddf5b6fc02b1ad0b889
```
which shows the dask name (hash) is unstable.

#### Expected Output
For processing large datasets, it's convenient to have a stable hash name in order to save intermediate results or compare them among developers. 
> Workaround: use `dask.array.mean()` instead of `dask.array.nanmean()`

Discussion:

Patch:
diff --git a/dask/array/core.py b/dask/array/core.py
index 9346b1c04..8801c75aa 100644
--- a/dask/array/core.py
+++ b/dask/array/core.py
@@ -2159,7 +2159,9 @@ def atop(func, out_ind, *args, **kwargs):
     concatenate : bool, keyword only
         If true concatenate arrays along dummy indices, else provide lists
     adjust_chunks : dict
-        Dictionary mapping index to function to be applied to chunk sizes
+        Dictionary mapping index to information to adjust chunk sizes.  Can
+        either be a constant chunksize, a tuple of all chunksizes, or a
+        function that converts old chunksize to new chunksize
     new_axes : dict, keyword only
         New indexes and their dimension lengths
 
@@ -2568,24 +2570,73 @@ def concatenate(seq, axis=0, allow_unknown_chunksizes=False):
     return Array(dsk2, name, chunks, dtype=dt)
 
 
-def insert_to_ooc(arr, out, lock=True, region=None):
-    if lock is True:
-        lock = Lock()
+def store_chunk(x, out, index, lock, region):
+    """
+    A function inserted in a Dask graph for storing a chunk.
 
-    def store(x, out, index, lock, region):
-        subindex = index
-        if region is not None:
-            subindex = fuse_slice(region, index)
+    Parameters
+    ----------
+    x: array-like
+        An array (potentially a NumPy one)
+    out: array-like
+        Where to store results too.
+    index: slice-like
+        Where to store result from ``x`` in ``out``.
+    lock: Lock-like or False
+        Lock to use before writing to ``out``.
+    region: slice-like or None
+        Where relative to ``out`` to store ``x``.
+
+    Examples
+    --------
 
+    >>> a = np.ones((5, 6))
+    >>> b = np.empty(a.shape)
+    >>> store_chunk(a, b, (slice(None), slice(None)), False, None)
+    """
+
+    subindex = index
+    if region is not None:
+        subindex = fuse_slice(region, index)
+
+    if lock:
+        lock.acquire()
+    try:
+        out[subindex] = np.asanyarray(x)
+    finally:
         if lock:
-            lock.acquire()
-        try:
-            out[subindex] = np.asanyarray(x)
-        finally:
-            if lock:
-                lock.release()
+            lock.release()
+
+    return None
+
+
+def insert_to_ooc(arr, out, lock=True, region=None):
+    """
+    Creates a Dask graph for storing chunks from ``arr`` in ``out``.
+
+    Parameters
+    ----------
+    arr: da.Array
+        A dask array
+    out: array-like
+        Where to store results too.
+    lock: Lock-like or bool, optional
+        Whether to lock or with what (default is ``True``,
+        which means a ``threading.Lock`` instance).
+    region: slice-like, optional
+        Where in ``out`` to store ``arr``'s results
+        (default is ``None``, meaning all of ``out``).
+
+    Examples
+    --------
+    >>> import dask.array as da
+    >>> d = da.ones((5, 6), chunks=(2, 3))
+    >>> a = np.empty(d.shape)
+    >>> insert_to_ooc(d, a)  # doctest: +SKIP
+    """
 
-        return None
+    if lock is True:
+        lock = Lock()
 
     slices = slices_from_chunks(arr.chunks)
 
@@ -2593,7 +2644,9 @@ def insert_to_ooc(arr, out, lock=True, region=None):
     dsk = dict()
     for t, slc in zip(core.flatten(arr.__dask_keys__()), slices):
         store_key = (name,) + t[1:]
-        dsk[store_key] = (store, t, out, slc, lock, region)
+        dsk[store_key] = (
+            store_chunk, t, out, slc, lock, region
+        )
 
     return dsk
 
diff --git a/dask/array/percentile.py b/dask/array/percentile.py
index 6368231a5..e86894d6a 100644
--- a/dask/array/percentile.py
+++ b/dask/array/percentile.py
@@ -2,6 +2,7 @@ from __future__ import absolute_import, division, print_function
 
 from functools import wraps
 from collections import Iterator
+from numbers import Number
 
 import numpy as np
 from toolz import merge, merge_sorted
@@ -38,6 +39,8 @@ def percentile(a, q, interpolation='linear'):
     if not a.ndim == 1:
         raise NotImplementedError(
             "Percentiles only implemented for 1-d arrays")
+    if isinstance(q, Number):
+        q = [q]
     q = np.array(q)
     token = tokenize(a, list(q), interpolation)
     name = 'percentile_chunk-' + token
diff --git a/dask/bag/core.py b/dask/bag/core.py
index 4428e0ba5..62297b290 100644
--- a/dask/bag/core.py
+++ b/dask/bag/core.py
@@ -1,16 +1,15 @@
 from __future__ import absolute_import, division, print_function
 
-from collections import Iterable, Iterator, defaultdict
-from functools import wraps, partial
 import itertools
 import math
-from operator import getitem
 import types
 import uuid
-from random import Random
-from warnings import warn
+import warnings
+from collections import Iterable, Iterator, defaultdict
 from distutils.version import LooseVersion
-
+from functools import wraps, partial
+from operator import getitem
+from random import Random
 
 from toolz import (merge, take, reduce, valmap, map, partition_all, filter,
                    remove, compose, curry, first, second, accumulate, peek)
@@ -1050,7 +1049,7 @@ class Bag(Base):
 
         return type(self)(merge(self.dask, dsk), e, 1)
 
-    def take(self, k, npartitions=1, compute=True):
+    def take(self, k, npartitions=1, compute=True, warn=True):
         """ Take the first k elements.
 
         Parameters
@@ -1064,6 +1063,9 @@ class Bag(Base):
             returned. Pass -1 to use all partitions.
         compute : bool, optional
             Whether to compute the result, default is True.
+        warn : bool, optional
+            Whether to warn if the number of elements returned is less than
+            requested, default is True.
 
         >>> b = from_sequence(range(10))
         >>> b.take(3)  # doctest: +SKIP
@@ -1087,9 +1089,9 @@ class Bag(Base):
                 dsk[(name_p, i)] = (list, (take, k, (self.name, i)))
 
             concat = (toolz.concat, ([(name_p, i) for i in range(npartitions)]))
-            dsk[(name, 0)] = (safe_take, k, concat)
+            dsk[(name, 0)] = (safe_take, k, concat, warn)
         else:
-            dsk = {(name, 0): (safe_take, k, (self.name, 0))}
+            dsk = {(name, 0): (safe_take, k, (self.name, 0), warn)}
 
         b = Bag(merge(self.dask, dsk), name, 1)
 
@@ -1215,12 +1217,16 @@ class Bag(Base):
         import dask.dataframe as dd
         if meta is None:
             if isinstance(columns, pd.DataFrame):
-                warn("Passing metadata to `columns` is deprecated. Please "
-                     "use the `meta` keyword instead.")
+                warnings.warn("Passing metadata to `columns` is deprecated. "
+                              "Please use the `meta` keyword instead.")
                 meta = columns
             else:
-                head = self.take(1)[0]
-                meta = pd.DataFrame([head], columns=columns)
+                head = self.take(1, warn=False)
+                if len(head) == 0:
+                    raise ValueError("`dask.bag.Bag.to_dataframe` failed to "
+                                     "properly infer metadata, please pass in "
+                                     "metadata via the `meta` keyword")
+                meta = pd.DataFrame(list(head), columns=columns)
         elif columns is not None:
             raise ValueError("Can't specify both `meta` and `columns`")
         else:
@@ -1986,12 +1992,12 @@ def empty_safe_aggregate(func, parts, is_last):
     return empty_safe_apply(func, parts2, is_last)
 
 
-def safe_take(n, b):
+def safe_take(n, b, warn=True):
     r = list(take(n, b))
-    if len(r) != n:
-        warn("Insufficient elements for `take`. {0} elements requested, "
-             "only {1} elements available. Try passing larger `npartitions` "
-             "to `take`.".format(n, len(r)))
+    if len(r) != n and warn:
+        warnings.warn("Insufficient elements for `take`. {0} elements "
+                      "requested, only {1} elements available. Try passing "
+                      "larger `npartitions` to `take`.".format(n, len(r)))
     return r
 
 
diff --git a/dask/base.py b/dask/base.py
index 88777b0ab..f13786387 100644
--- a/dask/base.py
+++ b/dask/base.py
@@ -584,8 +584,10 @@ def _normalize_function(func):
         funcs = reversed((first,) + func.funcs) if first else func.funcs
         return tuple(normalize_function(f) for f in funcs)
     elif isinstance(func, partial):
-        kws = tuple(sorted(func.keywords.items())) if func.keywords else ()
-        return (normalize_function(func.func), func.args, kws)
+        args = tuple(normalize_token(i) for i in func.args)
+        kws = tuple((k, normalize_token(v))
+                    for k, v in sorted(func.keywords.items()))
+        return (normalize_function(func.func), args, kws)
     else:
         try:
             result = pickle.dumps(func, protocol=0)
diff --git a/dask/dataframe/__init__.py b/dask/dataframe/__init__.py
index 1dabbe714..0e954e44a 100644
--- a/dask/dataframe/__init__.py
+++ b/dask/dataframe/__init__.py
@@ -9,10 +9,7 @@ from .io import (from_array, from_pandas, from_bcolz,
                  demo, to_hdf, to_records, to_bag)
 from .optimize import optimize
 from .multi import merge, concat
-from .rolling import (rolling_count, rolling_sum, rolling_mean, rolling_median,
-                      rolling_min, rolling_max, rolling_std, rolling_var,
-                      rolling_skew, rolling_kurt, rolling_quantile, rolling_apply,
-                      rolling_window)
+from . import rolling
 from ..base import compute
 from .reshape import get_dummies, pivot_table, melt
 try:
diff --git a/dask/dataframe/rolling.py b/dask/dataframe/rolling.py
index e6e606698..501ac6fa8 100644
--- a/dask/dataframe/rolling.py
+++ b/dask/dataframe/rolling.py
@@ -1,8 +1,6 @@
 from __future__ import absolute_import, division, print_function
 
 import datetime
-import warnings
-from functools import wraps
 
 import pandas as pd
 from pandas.core.window import Rolling as pd_Rolling
@@ -142,27 +140,6 @@ def map_overlap(func, df, before, after, *args, **kwargs):
     return df._constructor(dsk, name, meta, df.divisions)
 
 
-def wrap_rolling(func, method_name):
-    """Create a chunked version of a pandas.rolling_* function"""
-    @wraps(func)
-    def rolling(arg, window, *args, **kwargs):
-        # pd.rolling_* functions are deprecated
-        warnings.warn(("DeprecationWarning: dd.rolling_{0} is deprecated and "
-                       "will be removed in a future version, replace with "
-                       "df.rolling(...).{0}(...)").format(method_name))
-
-        rolling_kwargs = {}
-        method_kwargs = {}
-        for k, v in kwargs.items():
-            if k in {'min_periods', 'center', 'win_type', 'axis', 'freq'}:
-                rolling_kwargs[k] = v
-            else:
-                method_kwargs[k] = v
-        rolling = arg.rolling(window, **rolling_kwargs)
-        return getattr(rolling, method_name)(*args, **method_kwargs)
-    return rolling
-
-
 def _head_timedelta(current, next_, after):
     """Return rows of ``next_`` whose index is before the last
     observation in ``current`` + ``after``.
@@ -197,27 +174,6 @@ def _tail_timedelta(prev, current, before):
     return prev[prev.index > (current.index.min() - before)]
 
 
-rolling_count = wrap_rolling(pd.rolling_count, 'count')
-rolling_sum = wrap_rolling(pd.rolling_sum, 'sum')
-rolling_mean = wrap_rolling(pd.rolling_mean, 'mean')
-rolling_median = wrap_rolling(pd.rolling_median, 'median')
-rolling_min = wrap_rolling(pd.rolling_min, 'min')
-rolling_max = wrap_rolling(pd.rolling_max, 'max')
-rolling_std = wrap_rolling(pd.rolling_std, 'std')
-rolling_var = wrap_rolling(pd.rolling_var, 'var')
-rolling_skew = wrap_rolling(pd.rolling_skew, 'skew')
-rolling_kurt = wrap_rolling(pd.rolling_kurt, 'kurt')
-rolling_quantile = wrap_rolling(pd.rolling_quantile, 'quantile')
-rolling_apply = wrap_rolling(pd.rolling_apply, 'apply')
-
-
-@wraps(pd.rolling_window)
-def rolling_window(arg, window, **kwargs):
-    if kwargs.pop('mean', True):
-        return rolling_mean(arg, window, **kwargs)
-    return rolling_sum(arg, window, **kwargs)
-
-
 def pandas_rolling_method(df, rolling_kwargs, name, *args, **kwargs):
     rolling = df.rolling(**rolling_kwargs)
     return getattr(rolling, name)(*args, **kwargs)
diff --git a/dask/dot.py b/dask/dot.py
index 33f809911..e5fedfbd9 100644
--- a/dask/dot.py
+++ b/dask/dot.py
@@ -125,23 +125,29 @@ def to_graphviz(dsk, data_attributes=None, function_attributes=None,
         k_name = name(k)
         if k_name not in seen:
             seen.add(k_name)
-            g.node(k_name, label=label(k, cache=cache), shape='box',
-                   **data_attributes.get(k, {}))
+            attrs = data_attributes.get(k, {})
+            attrs.setdefault('label', label(k, cache=cache))
+            attrs.setdefault('shape', 'box')
+            g.node(k_name, **attrs)
 
         if istask(v):
             func_name = name((k, 'function'))
             if func_name not in seen:
                 seen.add(func_name)
-                g.node(func_name, label=task_label(v), shape='circle',
-                       **function_attributes.get(k, {}))
+                attrs = function_attributes.get(k, {})
+                attrs.setdefault('label', task_label(v))
+                attrs.setdefault('shape', 'circle')
+                g.node(func_name, **attrs)
             g.edge(func_name, k_name)
 
             for dep in get_dependencies(dsk, k):
                 dep_name = name(dep)
                 if dep_name not in seen:
                     seen.add(dep_name)
-                    g.node(dep_name, label=label(dep, cache=cache), shape='box',
-                           **data_attributes.get(dep, {}))
+                    attrs = data_attributes.get(dep, {})
+                    attrs.setdefault('label', label(dep, cache=cache))
+                    attrs.setdefault('shape', 'box')
+                    g.node(dep_name, **attrs)
                 g.edge(dep_name, func_name)
         elif ishashable(v) and v in dsk:
             g.edge(name(v), k_name)
diff --git a/docs/source/changelog.rst b/docs/source/changelog.rst
index ada3406ce..e7bed3642 100644
--- a/docs/source/changelog.rst
+++ b/docs/source/changelog.rst
@@ -7,6 +7,7 @@ Changelog
 Array
 +++++
 
+- Fix handling of scalar percentile values in ``percentile`` (:pr:`3021`) `James Bourbeau`_
 - Prevent ``bool()`` coercion from calling compute (:pr:`2958`) `Albert DeFusco`_
 - Add ``matmul`` (:pr:`2904`) `John A Kirkham`_
 - Support N-D arrays with ``matmul`` (:pr:`2909`) `John A Kirkham`_
@@ -32,6 +33,7 @@ DataFrame
 - Correctly handle the column name (`df.columns.name`) when reading in ``dd.read_parquet`` (:pr:2973`) `Tom Augspurger`_
 - Fixed ``dd.concat`` losing the index dtype when the data contained a categorical (:issue:`2932`) `Tom Augspurger`_
 - ``DataFrame.merge()`` (:pr:`2960`) now supports merging on a combination of columns and the index `Jon Mease`_
+- Removed the deprecated ``dd.rolling*`` methods, in preperation for their removal in the next pandas release (:pr:`2995`) `Tom Augspurger`_
 
 
 Core
diff --git a/docs/source/dataframe-api.rst b/docs/source/dataframe-api.rst
index b39ca00dc..a28df37c6 100644
--- a/docs/source/dataframe-api.rst
+++ b/docs/source/dataframe-api.rst
@@ -243,19 +243,25 @@ Rolling Operations
 
 .. autosummary::
    rolling.map_overlap
-   rolling.rolling_apply
-   rolling.rolling_count
-   rolling.rolling_kurt
-   rolling.rolling_max
-   rolling.rolling_mean
-   rolling.rolling_median
-   rolling.rolling_min
-   rolling.rolling_quantile
-   rolling.rolling_skew
-   rolling.rolling_std
-   rolling.rolling_sum
-   rolling.rolling_var
-   rolling.rolling_window
+   Series.rolling
+   DataFrame.rolling
+
+.. currentmodule:: dask.dataframe.rolling
+
+.. autosummary::
+   Rolling.apply
+   Rolling.count
+   Rolling.kurt
+   Rolling.max
+   Rolling.mean
+   Rolling.median
+   Rolling.min
+   Rolling.quantile
+   Rolling.skew
+   Rolling.std
+   Rolling.sum
+   Rolling.var
+
 
 Create DataFrames
 ~~~~~~~~~~~~~~~~~
@@ -346,20 +352,7 @@ Rolling
 
 .. currentmodule:: dask.dataframe.rolling
 
-.. autofunction:: rolling_apply
 .. autofunction:: map_overlap
-.. autofunction:: rolling_count
-.. autofunction:: rolling_kurt
-.. autofunction:: rolling_max
-.. autofunction:: rolling_mean
-.. autofunction:: rolling_median
-.. autofunction:: rolling_min
-.. autofunction:: rolling_quantile
-.. autofunction:: rolling_skew
-.. autofunction:: rolling_std
-.. autofunction:: rolling_sum
-.. autofunction:: rolling_var
-.. autofunction:: rolling_window
 
 
 Other functions
diff --git a/docs/source/inspect.rst b/docs/source/inspect.rst
index 2cd98f57a..4725af80a 100644
--- a/docs/source/inspect.rst
+++ b/docs/source/inspect.rst
@@ -19,7 +19,7 @@ The first step is to look at the ``.dask`` attribute of an array
 
    >>> import dask.array as da
    >>> x = da.ones((5, 15), chunks=(5, 5))
-   >>> x.dask
+   >>> dict(x.dask)
    {('wrapped_1', 0, 0): (ones, (5, 5)),
     ('wrapped_1', 0, 1): (ones, (5, 5)),
     ('wrapped_1', 0, 2): (ones, (5, 5))}
@@ -29,7 +29,7 @@ objects
 
 .. code-block:: python
 
-   >>> (x + 1).dask
+   >>> dict((x + 1).dask)
    {('wrapped_1', 0, 0): (ones, (5, 5)),
     ('wrapped_1', 0, 1): (ones, (5, 5)),
     ('wrapped_1', 0, 2): (ones, (5, 5))
@@ -37,6 +37,9 @@ objects
     ('x_1', 0, 1): (add, ('wrapped_1', 0, 1), 1),
     ('x_1', 0, 2): (add, ('wrapped_1', 0, 2), 1)}
 
+.. note:: In this example we use simple names like ``x_1``, ``ones``, and
+  ``add`` for demonstration purposes.  However in practice these names may be
+  more complex and include long hashed names.
 
 Visualize graphs with DOT
 -------------------------
diff --git a/docs/source/setup/kubernetes.rst b/docs/source/setup/kubernetes.rst
index bb75e2d25..1b877f65e 100644
--- a/docs/source/setup/kubernetes.rst
+++ b/docs/source/setup/kubernetes.rst
@@ -136,6 +136,7 @@ pip packages to install on the both the workers and Jupyter server (these two
 environments should be matched).
 
 .. code-block:: yaml
+
    # config.yaml
 
    worker:
diff --git a/docs/source/spark.rst b/docs/source/spark.rst
index c5a693974..a0997c10d 100644
--- a/docs/source/spark.rst
+++ b/docs/source/spark.rst
@@ -1,13 +1,10 @@
-Comparison to PySpark
-=====================
+Comparison to Spark
+===================
 
-Spark_ is a popular distributed computing tool with a decent Python API
-PySpark_.  Spark is growing to become a dominant name today in Big Data
-analysis alongside Hadoop, for which MRJob_ is possibly the dominant
-Python layer.
-
-Dask has several elements that appear to intersect this space and we are often
-asked, "How does Dask compare with Spark?"
+`Apache Spark <https://spark.apache.org/>`_ is a popular distributed computing
+tool for tabular datasets that is growing to become a dominant name in Big Data
+analysis today.  Dask has several elements that appear to intersect this space
+and we are often asked, "How does Dask compare with Spark?"
 
 Answering such comparison questions in an unbiased and informed way is hard,
 particularly when the differences can be somewhat technical.  This document
@@ -16,103 +13,114 @@ tries to do this; we welcome any corrections.
 Summary
 -------
 
-Apache Spark is an all-inclusive framework combining distributed computing,
-SQL queries, machine learning, and more that runs on the JVM and is commonly
-co-deployed with other Big Data frameworks like Hadoop.  It was originally
-optimized for bulk data ingest and querying common in data engineering and
-business analytics but has since broadened out.  Spark is typically used on
-small to medium sized cluster but also runs well on a single machine.
-
-Dask is a parallel programming library that combines with the Numeric Python
-ecosystem to provide parallel arrays, dataframes, machine learning, and custom
-algorithms.  It is based on Python and the foundational C/Fortran stack.  Dask
-was originally designed to complement other libraries with parallelism,
-particular for numeric computing and advanced analytics, but has since
-broadened out.  Dask is typically used on a single machine, but also runs well
-on a distributed cluster.
-
 Generally Dask is smaller and lighter weight than Spark.  This means that it
 has fewer features and instead is intended to be used in conjunction with other
-libraries, particularly those in the numeric Python ecosystem.
-
-
-User-Facing Differences
------------------------
-
-Scale
-~~~~~
-
-Spark began its life aimed at the thousand node cluster case.  As
-such it thinks well about worker failures and integration with data-local
-file systems like the Hadoop FileSystem (HDFS).  That being said, Spark can
-run in standalone mode on a single machine.
-
-Dask began its life building out parallel algorithms for numerical array
-computations on a single computer.  As such it thinks well about low-latency
-scheduling, low memory footprints, shared memory, and efficient use of local
-disk.  That being said dask can run on a distributed_ cluster, making use of
-HDFS and other Big Data technologies.
-
-.. _distributed: https://distributed.readthedocs.io/
-
-
-Java Python Performance
-~~~~~~~~~~~~~~~~~~~~~~~
-
-Spark is written in Scala, a multi-paradigm language built on top of the Java
-Virtual Machine (JVM).  Since the rise of Hadoop, Java based languages have
-steadily gained traction on data warehousing tasks and are good at managing
-large amounts of heterogeneous data such as you might find in JSON blobs.  The
-Spark development team is now focusing more on binary and native data formats
-with their new effort, Tungsten.
-
-Dask is written in Python, a multi-paradigm language built on top of the
-C/Fortran native stack.  This stack benefits from decades of scientific research
-optimizing very fast computation on numeric data.  As such, dask is already
-very good on analytic computations on data such as you might find in HDF5 files
-or analytic databases.  It can also handle JSON blob type data using Python
-data structures (which are `surprisingly fast`_) using the cytoolz_ library in
-parallel.
-
-
-Java Python Disconnect
-~~~~~~~~~~~~~~~~~~~~~~
-
-Python users on Spark sometimes express frustration by how far separated they
-are from computations.  Some of this is inevitable; distributed debugging is a
-hard problem.  Some of it however is due to having to hop over the JVM.  Spark
-workers spin up JVMs which in turn spin up Python processes.  Data moving back
-and forth makes extra trips both through a distributed cluster and also through
-extra serialization layers (see py4j_) and computation layers.  Limitations
-like the Java heap size and large Java stack traces come as a surprise to users
-accustomed to native code execution.
-
-Dask has an advantage for Python users because it is itself a Python library,
-so serialization and debugging when things go wrong happens more smoothly.
-
-However, Dask only benefits Python users while Spark is useful in a
-variety of JVM languages (Scala, Java, Clojure) and also has limited support in
-Python and R.  New Spark projects like the DataFrame skip serialization and
-boxed execution issues by forgoing the Python process entirely and instead have
-Python code drive native Scala code.  APIs for these libraries tend to lag a
-bit behind their Scala counterparts.
-
-
-Scope
-~~~~~
-
-Spark was originally built around the RDD, an unordered collection allowing
-repeats.  Most spark add-ons were built on top of this construct, inheriting
-both its abilities and limitations.
-
-Dask is built on a lower-level and more general construct of a generic task
-graph with arbitrary data dependencies.  This allows more general computations
-to be built by users within the dask framework.  This is probably the largest
-fundamental difference between the two projects.  Dask gives up high-level
-understanding to allow users to express more complex parallel algorithms.  This
-ended up being essential when writing complex projects like ``dask.array``,
-datetime algorithms in ``dask.dataframe`` or non-trivial algorithms in machine
-learning.
+libraries, particularly those in the numeric Python ecosystem.  It couples with
+other libraries like Pandas or Scikit-Learn to achieve high-level
+functionality.
+
+- Language
+    - Spark is written in Scala with some support for Python and R.  It
+      interoperates well with other JVM code.
+    - Dask is written in Python and only really supports Python.  It
+      interoperates well with C/C++/Fortran/LLVM or other natively compiled
+      code linked through Python.
+- Ecosystem
+    - Spark is an all-in-one project that has inspired its own ecosystem.  It
+      integrates well with many other Apache projects.
+    - Dask is a component of the larger Python ecosystem.  It couples with and
+      enhances other libraries like NumPy, Pandas, and Scikit-Learn.
+- Age and Trust
+    - Spark is older (since 2010) and has become a dominant and
+      well-trusted tool in the Big Data enterprise world.
+    - Dask is younger (since 2014) and is an extension of the
+      well trusted NumPy/Pandas/Scikit-learn/Jupyter stack.
+- Scope
+    - Spark is more focused on traditional business intelligence
+      operations like SQL and lightweight machine learning.
+    - Dask is applied more generally both to business intelligence
+      applications, as well as a number of scientific and custom business
+      situations
+- Internal Design
+    - Spark's internal model is higher level, providing good high level
+      optimizations on uniformly applied computations, but lacking flexibility
+      for more complex algorithms or ad-hoc systems.  It is fundamentally an
+      extension of the Map-Shuffle-Reduce paradigm.
+    - Dask's internal model is lower level, and so lacks high level
+      optimizations, but is able to implement more sophisticated algorithms and
+      build more complex bespoke systems.  It is fundamentally based on generic
+      task scheduling.
+- Scale
+    - Spark scales from a single node to thousand-node clusters
+    - Dask scales from a single node to thousand-node clusters
+- APIs
+    - Dataframes
+        - Spark dataframe has its own API and memory model.  It also
+          implements a large subset of the SQL language.  Spark includes a
+          high-level query optimizer for complex queries.
+        - Dask.dataframe reuses the Pandas API and memory model.  It implements
+          neither SQL nor a query optimizer.  It is able to do random access,
+          efficient time series operations, and other Pandas-style indexed
+          operations.
+    - Machine Learning
+        - Spark MLLib is a cohesive project with support for common operations
+          that are easy to implement with Spark's Map-Shuffle-Reduce style
+          system.  People considering MLLib might also want to consider *other*
+          JVM-based machine learning libraries like H2O, which may have better
+          performance.
+        - Dask relies on and interoperates with existing libraries like
+          Scikit-Learn and XGBoost.  These can be more familiar or higher
+          performance, but generally results in a less-cohesive whole.  See the
+          `dask-ml`_ project for integrations.
+    - Arrays
+        - Spark does not include support for multi-dimensional arrays natively
+          (this would be challenging given their computation model) although
+          some support for two-dimensional matrices may be found in MLLib.
+          People may also want to look at the `Thunder
+          <https://github.com/thunder-project/thunder>`_ project.
+        - Dask fully supports the NumPy model for :doc:`scalable multi-dimensional
+          arrays <array>`.
+    - Streaming
+        - Spark's support for streaming data is first-class and integrates well
+          into their other APIs.  It follows a mini-batch approach.  This
+          provides decent performance on large uniform streaming operations
+        - Dask provides a real-time futures interface that is lower-level than
+          Spark streaming.  This enables more creative and complex use-cases,
+          but requires more work than Spark streaming.
+    - Graphs / complex networks
+        - Spark provides GraphX, a library for graph processing
+        - Dask provides no such library
+    - Custom parallelism
+        - Spark generally expects users to compose computations out of their
+          high-level primitives (map, reduce, groupby, join, ...).  It is also
+          possible to extend Spark through subclassing RDDs, although this is
+          rarely done.
+        - Dask allows you to specify arbitrary task graphs for more complex and
+          custom systems that are not part of the standard set of collections.
+
+.. _dask-ml: https://dask-ml.readthedocs.org/en/latest
+
+
+Reasons you might choose Spark
+------------------------------
+
+-  You prefer Scala or the SQL language
+-  You have mostly JVM infrastructure and legacy systems
+-  You want an established and trusted solution for business
+-  You are mostly doing business analytics with some lightweight machine learning
+-  You want an all-in-one solution
+
+
+Reasons you might choose Dask
+-----------------------------
+
+-  You prefer Python or native code, or have large legacy code bases that you
+   do not want to entirely rewrite
+-  Your use case is complex or does not cleanly fit the Spark computing model
+-  You want a lighter-weight transition from single-machine computing to
+   cluster computing
+-  You want to interoperate with other technologies and don't mind installing
+   multiple packages
 
 
 Developer-Facing Differences
@@ -142,10 +150,6 @@ can because Dask schedulers do not have a top-down picture of the computation
 they were asked to perform.  However, dask is able to easily represent far more
 `complex algorithms`_ and expose the creation of these algorithms to normal users.
 
-Dask.bag, the equivalent of the Spark.RDD, is just one abstraction built on top
-of dask.  Others exist.  Alternatively power-users can forego high-level
-collections entirely and jump straight to direct low-level task scheduling.
-
 
 Coding Styles
 ~~~~~~~~~~~~~
@@ -174,11 +178,6 @@ then you should forget both Spark and Dask and use Postgres_ or MongoDB_.
 
 .. _Spark: https://spark.apache.org/
 .. _PySpark: https://spark.apache.org/docs/latest/api/python/
-.. _Hadoop: https://hadoop.apache.org/
-.. _MRJob: https://mrjob.readthedocs.io
-.. _`surprisingly fast`: https://www.youtube.com/watch?v=PpBK4zIaFLE
-.. _cytoolz: https://toolz.readthedocs.io
-.. _py4j: http://py4j.sourceforge.net/
 .. _Postgres: http://www.postgresql.org/
 .. _MongoDB: https://www.mongodb.org/
 .. _`complex algorithms`: http://matthewrocklin.com/blog/work/2015/06/26/Complex-Graphs


Test patch:
diff --git a/dask/array/tests/test_percentiles.py b/dask/array/tests/test_percentiles.py
index ec9f57bf0..1ff01aa6d 100644
--- a/dask/array/tests/test_percentiles.py
+++ b/dask/array/tests/test_percentiles.py
@@ -51,3 +51,11 @@ def test_percentile_with_categoricals():
 def test_percentiles_with_empty_arrays():
     x = da.ones(10, chunks=((5, 0, 5),))
     assert_eq(da.percentile(x, [10, 50, 90]), np.array([1, 1, 1], dtype=x.dtype))
+
+
+@pytest.mark.parametrize('q', [5, 5.0, np.int64(5), np.float64(5)])
+def test_percentiles_with_scaler_percentile(q):
+    # Regression test to ensure da.percentile works with scalar percentiles
+    # See #3020
+    d = da.ones((16,), chunks=(4,))
+    assert_eq(da.percentile(d, q), np.array([1], dtype=d.dtype))
diff --git a/dask/bag/tests/test_bag.py b/dask/bag/tests/test_bag.py
index 46427f714..44b874d48 100644
--- a/dask/bag/tests/test_bag.py
+++ b/dask/bag/tests/test_bag.py
@@ -530,17 +530,23 @@ def test_take_npartitions():
         b.take(1, npartitions=5)
 
 
-@pytest.mark.skipif(sys.version_info[:2] == (3,3),
-                    reason="Python3.3 uses pytest2.7.2, w/o warns method")
 def test_take_npartitions_warn():
-    with pytest.warns(None):
-        b.take(100)
+    # Use single-threaded scheduler so warnings are properly captured in the
+    # same process
+    with dask.set_options(get=dask.get):
+        with pytest.warns(UserWarning):
+            b.take(100)
+
+        with pytest.warns(UserWarning):
+            b.take(7)
 
-    with pytest.warns(None):
-        b.take(7)
+        with pytest.warns(None) as rec:
+            b.take(7, npartitions=2)
+        assert len(rec) == 0
 
-    with pytest.warns(None):
-        b.take(7, npartitions=2)
+        with pytest.warns(None) as rec:
+            b.take(7, warn=False)
+        assert len(rec) == 0
 
 
 def test_map_is_lazy():
@@ -775,6 +781,11 @@ def test_to_dataframe():
     with pytest.raises(ValueError):
         b.to_dataframe(columns=['a', 'b'], meta=sol)
 
+    # Inference fails if empty first partition
+    b2 = b.filter(lambda x: x['a'] > 200)
+    with pytest.raises(ValueError):
+        b2.to_dataframe()
+
     # Single column
     b = b.pluck('a')
     sol = sol[['a']]
diff --git a/dask/dataframe/tests/test_rolling.py b/dask/dataframe/tests/test_rolling.py
index 5e6695811..5b34ee690 100644
--- a/dask/dataframe/tests/test_rolling.py
+++ b/dask/dataframe/tests/test_rolling.py
@@ -93,48 +93,6 @@ def mad(x):
     return np.fabs(x - x.mean()).mean()
 
 
-def rolling_functions_tests(p, d):
-    # Old-fashioned rolling API
-    with pytest.warns(FutureWarning):
-        assert_eq(pd.rolling_count(p, 3), dd.rolling_count(d, 3))
-        assert_eq(pd.rolling_sum(p, 3), dd.rolling_sum(d, 3))
-        assert_eq(pd.rolling_mean(p, 3), dd.rolling_mean(d, 3))
-        assert_eq(pd.rolling_median(p, 3), dd.rolling_median(d, 3))
-        assert_eq(pd.rolling_min(p, 3), dd.rolling_min(d, 3))
-        assert_eq(pd.rolling_max(p, 3), dd.rolling_max(d, 3))
-        assert_eq(pd.rolling_std(p, 3), dd.rolling_std(d, 3))
-        assert_eq(pd.rolling_var(p, 3), dd.rolling_var(d, 3))
-        # see note around test_rolling_dataframe for logic concerning precision
-        assert_eq(pd.rolling_skew(p, 3),
-                  dd.rolling_skew(d, 3), check_less_precise=True)
-        assert_eq(pd.rolling_kurt(p, 3),
-                  dd.rolling_kurt(d, 3), check_less_precise=True)
-        assert_eq(pd.rolling_quantile(p, 3, 0.5), dd.rolling_quantile(d, 3, 0.5))
-        assert_eq(pd.rolling_apply(p, 3, mad), dd.rolling_apply(d, 3, mad))
-        # Test with edge-case window sizes
-        assert_eq(pd.rolling_sum(p, 0), dd.rolling_sum(d, 0))
-        assert_eq(pd.rolling_sum(p, 1), dd.rolling_sum(d, 1))
-        # Test with kwargs
-        assert_eq(pd.rolling_sum(p, 3, min_periods=3),
-                  dd.rolling_sum(d, 3, min_periods=3))
-        pytest.importorskip("scipy")
-        assert_eq(pd.rolling_window(p, 3, win_type='boxcar'),
-                  dd.rolling_window(d, 3, win_type='boxcar'))
-
-
-def test_rolling_functions_series():
-    ts = pd.Series(np.random.randn(25).cumsum())
-    dts = dd.from_pandas(ts, 3)
-    rolling_functions_tests(ts, dts)
-
-
-def test_rolling_functions_dataframe():
-    df = pd.DataFrame({'a': np.random.randn(25).cumsum(),
-                       'b': np.random.randint(100, size=(25,))})
-    ddf = dd.from_pandas(df, 3)
-    rolling_functions_tests(df, ddf)
-
-
 rolling_method_args_check_less_precise = [
     ('count', (), False),
     ('sum', (), False),
diff --git a/dask/tests/test_base.py b/dask/tests/test_base.py
index dfe9e3093..8ab1a76dc 100644
--- a/dask/tests/test_base.py
+++ b/dask/tests/test_base.py
@@ -32,16 +32,19 @@ np = import_or_none('numpy')
 pd = import_or_none('pandas')
 
 
-def test_normalize_function():
+def f1(a, b, c=1):
+    pass
+
 
-    def f1(a, b, c=1):
-        pass
+def f2(a, b=1, c=2):
+    pass
 
-    def f2(a, b=1, c=2):
-        pass
 
-    def f3(a):
-        pass
+def f3(a):
+    pass
+
+
+def test_normalize_function():
 
     assert normalize_function(f2)
 
@@ -168,6 +171,15 @@ def test_tokenize_numpy_ufunc_consistent():
     assert tokenize(inc) == tokenize(inc)
 
 
+def test_tokenize_partial_func_args_kwargs_consistent():
+    f = tz.partial(f3, f2, c=f1)
+    res = normalize_token(f)
+    sol = (b'cdask.tests.test_base\nf3\np0\n.',
+           (b'cdask.tests.test_base\nf2\np0\n.',),
+           (('c', b'cdask.tests.test_base\nf1\np0\n.'),))
+    assert res == sol
+
+
 def test_normalize_base():
     for i in [1, long(1), 1.1, '1', slice(1, 2, 3)]:
         assert normalize_token(i) is i
diff --git a/dask/tests/test_dot.py b/dask/tests/test_dot.py
index 1ec118669..06377a87e 100644
--- a/dask/tests/test_dot.py
+++ b/dask/tests/test_dot.py
@@ -19,7 +19,7 @@ from IPython.display import Image, SVG
 
 
 # Since graphviz doesn't store a graph, we need to parse the output
-label_re = re.compile('.*\[label=(.*?) shape=.*\]')
+label_re = re.compile('.*\[label=(.*?) shape=(.*?)\]')
 
 
 def get_label(line):
@@ -28,6 +28,12 @@ def get_label(line):
         return m.group(1)
 
 
+def get_shape(line):
+    m = label_re.match(line)
+    if m:
+        return m.group(2)
+
+
 dsk = {'a': 1,
        'b': 2,
        'c': (neg, 'a'),
@@ -73,6 +79,22 @@ def test_to_graphviz():
     funcs = set(('add', 'sum', 'neg'))
     assert set(labels).difference(dsk) == funcs
     assert set(labels).difference(funcs) == set(dsk)
+    shapes = list(filter(None, map(get_shape, g.body)))
+    assert set(shapes) == set(('box', 'circle'))
+
+
+def test_to_graphviz_custom():
+    g = to_graphviz(
+        dsk,
+        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)
+    shapes = list(filter(None, map(get_shape, g.body)))
+    assert set(shapes) == set(('box', 'circle', 'square', 'ellipse'))
 
 
 def test_to_graphviz_attributes():