Repository: dask/dask

Issue:
Series comparison to NumPy scalar fails
Comparing a dask Series with a NumPy scalar or 0D array fails in certain cases. Here is a minimal reproducible example:

```python
import dask.dataframe as dd
import pandas as pd

np.float64(5.2) >= dd.from_array(np.arange(10))
```

```
---------------------------------------------------------------------------
AttributeError                            Traceback (most recent call last)
<ipython-input-18-d905075c0565> in <module>()
      1 
----> 2 np.float64(10) >= dd.from_array(np.arange(10))

~/miniconda/envs/anacondaviz/lib/python3.6/site-packages/dask/dataframe/core.py in __array_wrap__(self, array, context)
   1704     def __array_wrap__(self, array, context=None):
   1705         if isinstance(context, tuple) and len(context) > 0:
-> 1706             index = context[1][0].index
   1707 
   1708         return pd.Series(array, index=index, name=self.name)

AttributeError: 'numpy.ndarray' object has no attribute 'index'
```

I can reproduce this using all releases in the dask 0.17.x series.

Discussion:
mrocklin: Thank you for taking this on @jrbourbeau !

@philippjfr does this resolve your original issue?
jrbourbeau: The original regression test I added was using the `dask.array.greater_equal` ufunc, which actually doesn't raise any errors. So I replaced the added test with the original code snippet given in issue #3392. 
mrocklin: It looks like there are some genuine testing errors in travis-ci.  I suspect that this fix works for some versions, but not others.
jrbourbeau: It looks like the NumPy 1.11.0 tests on Travis are failing because I've only modified `__array_ufunc__` (which is new in NumPy version 1.13) and not `__array_wrap__` (which is what's being used in the 1.11.0 tests).

Patch:
diff --git a/dask/array/__init__.py b/dask/array/__init__.py
index 67c697118..bf37a71b2 100644
--- a/dask/array/__init__.py
+++ b/dask/array/__init__.py
@@ -7,13 +7,14 @@ from .core import (Array, block, concatenate, stack, from_array, store,
                    broadcast_arrays, broadcast_to)
 from .routines import (take, choose, argwhere, where, coarsen, insert,
                        ravel, roll, unique, squeeze, ptp, diff, ediff1d,
-                       bincount, digitize, histogram, cov, array, dstack,
-                       vstack, hstack, compress, extract, round, count_nonzero,
-                       flatnonzero, nonzero, around, isin, isnull, notnull,
-                       isclose, allclose, corrcoef, swapaxes, tensordot,
-                       transpose, dot, vdot, matmul, apply_along_axis,
-                       apply_over_axes, result_type, atleast_1d, atleast_2d,
-                       atleast_3d, piecewise, flip, flipud, fliplr, einsum)
+                       gradient, bincount, digitize, histogram, cov, array,
+                       dstack, vstack, hstack, compress, extract, round,
+                       count_nonzero, flatnonzero, nonzero, around, isin,
+                       isnull, notnull, isclose, allclose, corrcoef, swapaxes,
+                       tensordot, transpose, dot, vdot, matmul,
+                       apply_along_axis, apply_over_axes, result_type,
+                       atleast_1d, atleast_2d, atleast_3d, piecewise, flip,
+                       flipud, fliplr, einsum)
 from .reshape import reshape
 from .ufunc import (add, subtract, multiply, divide, logaddexp, logaddexp2,
         true_divide, floor_divide, negative, power, remainder, mod, conj, exp,
diff --git a/dask/array/ghost.py b/dask/array/ghost.py
index 84538ef1b..c7be674df 100644
--- a/dask/array/ghost.py
+++ b/dask/array/ghost.py
@@ -361,7 +361,7 @@ def add_dummy_padding(x, depth, boundary):
     array([..., 0, 1, 2, 3, 4, 5, ...])
     """
     for k, v in boundary.items():
-        d = depth[k]
+        d = depth.get(k, 0)
         if v == 'none' and d > 0:
             empty_shape = list(x.shape)
             empty_shape[k] = d
@@ -465,4 +465,5 @@ def coerce_boundary(ndim, boundary):
         boundary = (boundary,) * ndim
     if isinstance(boundary, tuple):
         boundary = dict(zip(range(ndim), boundary))
+
     return boundary
diff --git a/dask/array/routines.py b/dask/array/routines.py
index 72e88433a..7c2c9f3ac 100644
--- a/dask/array/routines.py
+++ b/dask/array/routines.py
@@ -1,14 +1,15 @@
 from __future__ import division, print_function, absolute_import
 
 import inspect
+import math
 import warnings
 from collections import Iterable
 from distutils.version import LooseVersion
 from functools import wraps, partial
-from numbers import Integral
+from numbers import Number, Real, Integral
 
 import numpy as np
-from toolz import concat, sliding_window, interleave
+from toolz import concat, merge, sliding_window, interleave
 
 from .. import sharedict
 from ..core import flatten
@@ -404,6 +405,71 @@ def ediff1d(ary, to_end=None, to_begin=None):
     return r
 
 
+def _gradient_kernel(f, grad_varargs, grad_kwargs):
+    return np.gradient(f, *grad_varargs, **grad_kwargs)
+
+
+@wraps(np.gradient)
+def gradient(f, *varargs, **kwargs):
+    f = asarray(f)
+
+    if not all([isinstance(e, Number) for e in varargs]):
+        raise NotImplementedError("Only numeric scalar spacings supported.")
+
+    if varargs == ():
+        varargs = (1,)
+    if len(varargs) == 1:
+        varargs = f.ndim * varargs
+    if len(varargs) != f.ndim:
+        raise TypeError(
+            "Spacing must either be a scalar or a scalar per dimension."
+        )
+
+    kwargs["edge_order"] = math.ceil(kwargs.get("edge_order", 1))
+    if kwargs["edge_order"] > 2:
+        raise ValueError("edge_order must be less than or equal to 2.")
+
+    drop_result_list = False
+    axis = kwargs.pop("axis", None)
+    if axis is None:
+        axis = tuple(range(f.ndim))
+    elif isinstance(axis, Integral):
+        drop_result_list = True
+        axis = (axis,)
+
+    for e in axis:
+        if not isinstance(e, Integral):
+            raise TypeError("%s, invalid value for axis" % repr(e))
+        if not (-f.ndim <= e < f.ndim):
+            raise ValueError("axis, %s, is out of bounds" % repr(e))
+
+    if len(axis) != len(set(axis)):
+        raise ValueError("duplicate axes not allowed")
+
+    axis = tuple(ax % f.ndim for ax in axis)
+
+    if issubclass(f.dtype.type, (np.bool8, Integral)):
+        f = f.astype(float)
+    elif issubclass(f.dtype.type, Real) and f.dtype.itemsize < 4:
+        f = f.astype(float)
+
+    r = [
+        f.map_overlap(
+            _gradient_kernel,
+            dtype=f.dtype,
+            depth={j: 1 if j == ax else 0 for j in range(f.ndim)},
+            boundary="none",
+            grad_varargs=(varargs[i],),
+            grad_kwargs=merge(kwargs, {"axis": ax}),
+        )
+        for i, ax in enumerate(axis)
+    ]
+    if drop_result_list:
+        r = r[0]
+
+    return r
+
+
 @wraps(np.bincount)
 def bincount(x, weights=None, minlength=None):
     if minlength is None:
diff --git a/dask/dataframe/core.py b/dask/dataframe/core.py
index b00218d68..0ddbf62d4 100644
--- a/dask/dataframe/core.py
+++ b/dask/dataframe/core.py
@@ -344,8 +344,12 @@ class _Frame(DaskMethodsMixin, OperatorMethodMixin):
     def __array_ufunc__(self, numpy_ufunc, method, *inputs, **kwargs):
         out = kwargs.get('out', ())
         for x in inputs + out:
-            if not isinstance(x, (Number, Scalar, _Frame, Array,
-                                  pd.DataFrame, pd.Series, pd.Index)):
+            # ufuncs work with 0-dimensional NumPy ndarrays
+            # so we don't want to raise NotImplemented
+            if isinstance(x, np.ndarray) and x.shape == ():
+                continue
+            elif not isinstance(x, (Number, Scalar, _Frame, Array,
+                                    pd.DataFrame, pd.Series, pd.Index)):
                 return NotImplemented
 
         if method == '__call__':
@@ -1751,7 +1755,10 @@ class Series(_Frame):
 
     def __array_wrap__(self, array, context=None):
         if isinstance(context, tuple) and len(context) > 0:
-            index = context[1][0].index
+            if isinstance(context[1][0], np.ndarray) and context[1][0].shape == ():
+                index = None
+            else:
+                index = context[1][0].index
 
         return pd.Series(array, index=index, name=self.name)
 
@@ -2311,7 +2318,10 @@ class DataFrame(_Frame):
 
     def __array_wrap__(self, array, context=None):
         if isinstance(context, tuple) and len(context) > 0:
-            index = context[1][0].index
+            if isinstance(context[1][0], np.ndarray) and context[1][0].shape == ():
+                index = None
+            else:
+                index = context[1][0].index
 
         return pd.DataFrame(array, index=index, columns=self.columns)
 
diff --git a/dask/dataframe/groupby.py b/dask/dataframe/groupby.py
index 28a0854d8..ee086513e 100644
--- a/dask/dataframe/groupby.py
+++ b/dask/dataframe/groupby.py
@@ -1221,7 +1221,7 @@ class SeriesGroupBy(_GroupBy):
         if self._slice:
             result = result[self._slice]
 
-        if not isinstance(arg, (list, dict)):
+        if not isinstance(arg, (list, dict)) and isinstance(result, DataFrame):
             result = result[result.columns[0]]
 
         return result
diff --git a/dask/dataframe/io/demo.py b/dask/dataframe/io/demo.py
index 055059ede..54d62f3f2 100644
--- a/dask/dataframe/io/demo.py
+++ b/dask/dataframe/io/demo.py
@@ -51,7 +51,12 @@ def make_timeseries_part(start, end, dtypes, freq, state_data):
     return df
 
 
-def make_timeseries(start, end, dtypes, freq, partition_freq, seed=None):
+def make_timeseries(start='2000-01-01',
+                    end='2000-12-31',
+                    dtypes={'name': str, 'id': int, 'x': float, 'y': float},
+                    freq='10s',
+                    partition_freq='1M',
+                    seed=None):
     """ Create timeseries dataframe with random data
 
     Parameters
diff --git a/docs/source/array-api.rst b/docs/source/array-api.rst
index f6fa498c4..b0b77d5e7 100644
--- a/docs/source/array-api.rst
+++ b/docs/source/array-api.rst
@@ -83,6 +83,7 @@ Top level user functions:
    frompyfunc
    full
    full_like
+   gradient
    histogram
    hstack
    hypot
@@ -423,6 +424,7 @@ Other functions
 .. autofunction:: frompyfunc
 .. autofunction:: full
 .. autofunction:: full_like
+.. autofunction:: gradient
 .. autofunction:: histogram
 .. autofunction:: hstack
 .. autofunction:: hypot
diff --git a/docs/source/changelog.rst b/docs/source/changelog.rst
index 5ca12ab13..bfac52c44 100644
--- a/docs/source/changelog.rst
+++ b/docs/source/changelog.rst
@@ -18,11 +18,13 @@ Array
 - The ``topk`` API has changed from topk(k, array) to the more conventional topk(array, k).
   The legacy API still works but is now deprecated. (:pr:`2965`) `Guido Imperiale`_
 - New function ``argtopk`` for Dask Arrays (:pr:`3396`) `Guido Imperiale`_
+- Fix handling partial depth and boundary in ``map_overlap`` (:pr:`3445`) `John A Kirkham`_
+- Add ``gradient`` for Dask Arrays (:pr:`3434`) `John A Kirkham`_
+
 
 DataFrame
 +++++++++
 
-
 - Allow `t` as shorthand for `table` in `to_hdf` for pandas compatibility (:pr:`3330`) `Jörg Dietrich`_
 - Added top level `isna` method for Dask DataFrames (:pr:`3294`) `Christopher Ren`_
 - Fix selection on partition column on ``read_parquet`` for ``engine="pyarrow"`` (:pr:`3207`) `Uwe Korn`_
@@ -32,7 +34,10 @@ DataFrame
 - Provide more informative error message for meta= errors (:pr:`3343`) `Matthew Rocklin`_
 - add orc reader (:pr:`3284`) `Martin Durant`_
 - Default compression for parquet now always Snappy, in line with pandas (:pr:`3373`) `Martin Durant`_
+- Fixed bug in Dask DataFrame and Series comparisons with NumPy scalars (:pr:`3436`) `James Bourbeau`_
 - Remove outdated requirement from repartition docstring (:pr:`3440`) `Jörg Dietrich`_
+- Fixed bug in aggregation when only a Series is selected (:pr:`3446`) `Jörg Dietrich`_
+- Add default values to make_timeseries (:pr:`3421`) `Matthew Rocklin`_
 
 Bag
 +++
diff --git a/docs/source/conf.py b/docs/source/conf.py
index 548e5074f..8ac85a6a1 100644
--- a/docs/source/conf.py
+++ b/docs/source/conf.py
@@ -299,13 +299,6 @@ extlinks = {
 
 #  --Options for sphinx extensions -----------------------------------------------
 
-intersphinx_mapping = {'pandas': ('http://pandas.pydata.org/pandas-docs/stable/',
-                                  'http://pandas.pydata.org/pandas-docs/stable/objects.inv'),
-                       'numpy': ('https://docs.scipy.org/doc/numpy/',
-                                 'https://docs.scipy.org/doc/numpy/objects.inv')}
-
-#  --Options for sphinx extensions -----------------------------------------------
-
 intersphinx_mapping = {'pandas': ('http://pandas.pydata.org/pandas-docs/stable/',
                                   'http://pandas.pydata.org/pandas-docs/stable/objects.inv'),
                        'numpy': ('https://docs.scipy.org/doc/numpy/',
diff --git a/docs/source/futures.rst b/docs/source/futures.rst
index 838c13e83..4fb879f67 100644
--- a/docs/source/futures.rst
+++ b/docs/source/futures.rst
@@ -582,12 +582,14 @@ API
    fire_and_forget
    get_client
    secede
+   rejoin
    wait
 
 .. autofunction:: as_completed
 .. autofunction:: fire_and_forget
 .. autofunction:: get_client
 .. autofunction:: secede
+.. autofunction:: rejoin
 .. autofunction:: wait
 
 .. autoclass:: Client


Test patch:
diff --git a/dask/array/tests/test_ghost.py b/dask/array/tests/test_ghost.py
index 4aa770bde..0497ed95b 100644
--- a/dask/array/tests/test_ghost.py
+++ b/dask/array/tests/test_ghost.py
@@ -195,8 +195,11 @@ def test_map_overlap():
     exp1 = d.map_overlap(lambda x: x + x.size, depth=1, dtype=d.dtype)
     exp2 = d.map_overlap(lambda x: x + x.size, depth={0: 1, 1: 1},
                          boundary={0: 'reflect', 1: 'none'}, dtype=d.dtype)
+    exp3 = d.map_overlap(lambda x: x + x.size, depth={1: 1},
+                         boundary={1: 'reflect'}, dtype=d.dtype)
     assert_eq(exp1, x + 16)
     assert_eq(exp2, x + 12)
+    assert_eq(exp3, x + 8)
 
 
 @pytest.mark.parametrize("boundary", [
diff --git a/dask/array/tests/test_routines.py b/dask/array/tests/test_routines.py
index 7b8443644..a5a214eab 100644
--- a/dask/array/tests/test_routines.py
+++ b/dask/array/tests/test_routines.py
@@ -1,6 +1,7 @@
 from __future__ import division, print_function, absolute_import
 
 import itertools
+from numbers import Number
 import textwrap
 
 import pytest
@@ -419,6 +420,36 @@ def test_ediff1d(shape, to_end, to_begin):
     assert_eq(da.ediff1d(a, to_end, to_begin), np.ediff1d(x, to_end, to_begin))
 
 
+@pytest.mark.parametrize('shape, varargs, axis', [
+    [(10, 15, 20), (), None],
+    [(10, 15, 20), (2,), None],
+    [(10, 15, 20), (1.0, 1.5, 2.0), None],
+    [(10, 15, 20), (), 0],
+    [(10, 15, 20), (), 1],
+    [(10, 15, 20), (), 2],
+    [(10, 15, 20), (), -1],
+    [(10, 15, 20), (), (0, 2)],
+])
+@pytest.mark.parametrize('edge_order', [
+    1,
+    2
+])
+def test_gradient(shape, varargs, axis, edge_order):
+    a = np.random.randint(0, 10, shape)
+    d_a = da.from_array(a, chunks=(len(shape) * (5,)))
+
+    r = np.gradient(a, *varargs, axis=axis, edge_order=edge_order)
+    r_a = da.gradient(d_a, *varargs, axis=axis, edge_order=edge_order)
+
+    if isinstance(axis, Number):
+        assert_eq(r, r_a)
+    else:
+        assert len(r) == len(r_a)
+
+        for e_r, e_r_a in zip(r, r_a):
+            assert_eq(e_r, e_r_a)
+
+
 def test_bincount():
     x = np.array([2, 1, 5, 2, 1])
     d = da.from_array(x, chunks=2)
diff --git a/dask/bytes/tests/test_http.py b/dask/bytes/tests/test_http.py
index ce877c524..52ec7b991 100644
--- a/dask/bytes/tests/test_http.py
+++ b/dask/bytes/tests/test_http.py
@@ -2,6 +2,7 @@ import os
 import pytest
 import requests
 import subprocess
+import sys
 import time
 
 from dask.bytes.core import open_files
@@ -19,9 +20,9 @@ def dir_server():
                 f.write(b'a' * 10000)
 
         if PY2:
-            cmd = ['python', '-m', 'SimpleHTTPServer', '8999']
+            cmd = [sys.executable, '-m', 'SimpleHTTPServer', '8999']
         else:
-            cmd = ['python', '-m', 'http.server', '8999']
+            cmd = [sys.executable, '-m', 'http.server', '8999']
         p = subprocess.Popen(cmd, cwd=d)
         timeout = 10
         while True:
diff --git a/dask/dataframe/io/tests/test_demo.py b/dask/dataframe/io/tests/test_demo.py
index d6b71a270..03b43abb1 100644
--- a/dask/dataframe/io/tests/test_demo.py
+++ b/dask/dataframe/io/tests/test_demo.py
@@ -39,6 +39,13 @@ def test_make_timeseries():
     assert a._name != e._name
 
 
+def test_make_timeseries_no_args():
+    df = dd.demo.make_timeseries()
+    assert 1 < df.npartitions < 1000
+    assert len(df.columns) > 1
+    assert len(set(df.dtypes)) > 1
+
+
 def test_no_overlaps():
     df = dd.demo.make_timeseries('2000', '2001', {'A': float},
                                  freq='3H', partition_freq='3M')
diff --git a/dask/dataframe/tests/test_groupby.py b/dask/dataframe/tests/test_groupby.py
index 7d1f81d9b..370b54b56 100644
--- a/dask/dataframe/tests/test_groupby.py
+++ b/dask/dataframe/tests/test_groupby.py
@@ -1429,3 +1429,13 @@ def test_groupby_agg_custom__mode():
     expected = expected['cc'].groupby([expected['g0'], expected['g1']]).agg('sum')
 
     assert_eq(actual, expected)
+
+
+def test_groupby_select_column_agg():
+    pdf = pd.DataFrame({'A': [1, 2, 3, 1, 2, 3, 1, 2, 4],
+                        'B': [-0.776, -0.4, -0.873, 0.054, 1.419, -0.948,
+                              -0.967, -1.714, -0.666]})
+    ddf = dd.from_pandas(pdf, npartitions=4)
+    actual = ddf.groupby('A')['B'].agg('var')
+    expected = pdf.groupby('A')['B'].agg('var')
+    assert_eq(actual, expected)
diff --git a/dask/dataframe/tests/test_ufunc.py b/dask/dataframe/tests/test_ufunc.py
index 9e115ccd8..a83cc9fa4 100644
--- a/dask/dataframe/tests/test_ufunc.py
+++ b/dask/dataframe/tests/test_ufunc.py
@@ -374,3 +374,18 @@ def test_ufunc_with_reduction(redfunc, ufunc, pandas):
     with pytest.warns(None):
         assert isinstance(np_redfunc(dask), (dd.DataFrame, dd.Series, dd.core.Scalar))
         assert_eq(np_redfunc(np_ufunc(dask)), np_redfunc(np_ufunc(pandas)))
+
+
+@pytest.mark.parametrize('pandas',
+                         [pd.Series(np.random.randint(1, 100, size=100)),
+                          pd.DataFrame({'A': np.random.randint(1, 100, size=20),
+                                        'B': np.random.randint(1, 100, size=20),
+                                        'C': np.abs(np.random.randn(20))})])
+@pytest.mark.parametrize('scalar', [15, 16.4, np.int64(15), np.float64(16.4)])
+def test_ufunc_numpy_scalar_comparison(pandas, scalar):
+    # Regression test for issue #3392
+
+    dask_compare = scalar >= dd.from_pandas(pandas, npartitions=3)
+    pandas_compare = scalar >= pandas
+
+    assert_eq(dask_compare, pandas_compare)