dask / dask/dask

[ENH] Add support for cumulative aggregations in `GroupBy.agg`

Open
#9,620 1 comment 0 reactions 0 assignees View on GitHub
dataframe feature needs attention
Dominant language
Python
Stars
13.9k
Forks
2k
PR merge metrics
No merged PRs in 30d

Description

Currently, cumulative grouped aggregations can be called using their named methods, but passing them into `.agg()` raises an error:

```python
from dask.datasets import timeseries

ddf = timeseries()

ddf,groupby("name").cumsum() # works
ddf.groupby("name").agg("cumsum")
```

```python-traceback
ValueError Traceback (most recent call last)
Cell In [3], line 3
1 from dask.datasets import timeseries
----> 3 timeseries().groupby("name").agg("cumsum")

File ~/dev/dask/main/dask/dataframe/groupby.py:2557, in DataFrameGroupBy.agg(self, arg, split_every, split_out, shuffle, **kwargs)
2555 @_aggregate_docstring(based_on="pd.core.groupby.DataFrameGroupBy.agg")
2556 def agg(self, arg=None, split_every=None, split_out=1, shuffle=None, **kwargs):
-> 2557 return self.aggregate(
2558 arg=arg,
2559 split_every=split_every,
2560 split_out=split_out,
2561 shuffle=shuffle,
2562 **kwargs,
2563 )

File ~/dev/dask/main/dask/dataframe/groupby.py:2547, in DataFrameGroupBy.aggregate(self, arg, split_every, split_out, shuffle, **kwargs)
2544 if arg == "size":
2545 return self.size()
-> 2547 return super().aggregate(
2548 arg=arg,
2549 split_every=split_every,
2550 split_out=split_out,
2551 shuffle=shuffle,
2552 **kwargs,
2553 )

File ~/dev/dask/main/dask/dataframe/groupby.py:1976, in _GroupBy.aggregate(self, arg, split_every, split_out, shuffle, **kwargs)
1973 else:
1974 raise ValueError(f"aggregate on unknown object {self.obj}")
-> 1976 chunk_funcs, aggregate_funcs, finalizers = _build_agg_args(spec)
1978 if isinstance(self.by, (tuple, list)) and len(self.by) > 1:
1979 levels = list(range(len(self.by)))

File ~/dev/dask/main/dask/dataframe/groupby.py:833, in _build_agg_args(spec)
830 if not isinstance(func, Aggregation):
831 func = funcname(known_np_funcs.get(func, func))
--> 833 impls = _build_agg_args_single(result_column, func, input_column)
835 # overwrite existing result-columns, generate intermediates only once
836 for spec in impls["chunk_funcs"]:

File ~/dev/dask/main/dask/dataframe/groupby.py:886, in _build_agg_args_single(result_column, func, input_column)
883 return _build_agg_args_custom(result_column, func, input_column)
885 else:
--> 886 raise ValueError(f"unknown aggregate {func}")

ValueError: unknown aggregate cumsum
```

It looks like the underlying issue here is that we don't have a `_build_agg_args_*` function for any of the cumulative aggregations; not sure if that's all that needs to be done here to unblock this functionality.

Contributor guide

Open the contributing guide

Assessment

This issue has not been assessed yet.

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.