Repository: dask/dask

Issue:
Dataframe groupby()[column].agg fails with AttributeError
Minimal not working example (dask 0.17.2, pandas 0.22.0, Python 3.6):

```
import pandas as pd
from dask import dataframe as dd

df = 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(df, npartitions=1)
ddf.groupby('A')['B'].agg('var')
```

```
AttributeError                            Traceback (most recent call last)
<ipython-input-1-b43466d1ae6a> in <module>()
      6               -0.666]})
      7 ddf = dd.from_pandas(df, npartitions=1)
----> 8 ddf.groupby('A')['B'].agg('var')
      9 
     10 

~/applications/anaconda3/lib/python3.6/site-packages/dask/dataframe/groupby.py in agg(self, arg, split_every, split_out)
   1217     @derived_from(pd.core.groupby.SeriesGroupBy)
   1218     def agg(self, arg, split_every=None, split_out=1):
-> 1219         return self.aggregate(arg, split_every=split_every, split_out=split_out)

~/applications/anaconda3/lib/python3.6/site-packages/dask/dataframe/groupby.py in aggregate(self, arg, split_every, split_out)
   1211 
   1212         if not isinstance(arg, (list, dict)):
-> 1213             result = result[result.columns[0]]
   1214 
   1215         return result

AttributeError: 'Series' object has no attribute 'columns'
```

Discussion:

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/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/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..d74933916 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`_
@@ -33,6 +35,7 @@ DataFrame
 - add orc reader (:pr:`3284`) `Martin Durant`_
 - Default compression for parquet now always Snappy, in line with pandas (:pr:`3373`) `Martin Durant`_
 - 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`_
 
 Bag
 +++


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/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)