[ENH] Add support for cumulative aggregations in `GroupBy.agg`
- 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
Assessment
This issue has not been assessed yet.