Dask - groupby 期间行间时间戳差异的平均值

Question

我尝试在聚合期间使用 Dask 计算组行之间时间戳差异的平均值（两两）。

数据框的一个例子是：

   Timestamp            IP         Packets
0  2021-07-12 09:54:36  10.42.0.1  25
1  2021-07-12 09:54:36  10.41.0.1  5
2  2021-07-12 09:54:39  10.42.0.1  3
3  2021-07-12 09:54:46  10.41.0.1  30
4  2021-07-12 09:54:52  10.42.0.1  2

根据这个documentation，我尝试了这样的事情：

diffMean = dd.Aggregation('diffMean', lambda x: x.diff(), lambda x0: x0.mean().compute())
    
myDf = (df.groupby(['IP']).agg({ 'Packets': ['sum', 'mean'], 'Timestamp': ['min', 'max', diffMean]}).compute()) 

预期结果是：

           Packets    Timestamp
           sum  mean  min                  max                  diffMean
IP         
10.41.0.1  35   17.5  2021-07-12 09:54:36  2021-07-12 09:54:46  00:00:10 # 10 = 10 / 1
10.42.0.1  30   10    2021-07-12 09:54:36  2021-07-12 09:54:52  00:00:08 # 8 = (3 + 13) / 2

但是 Dask 不喜欢我的聚合...它给了我以下错误：

...
The above exception was the direct cause of the following exception:

Traceback (most recent call last):
  File "/home/flip/bin/resSys", line 185, in <module>
    main(sys.argv[1:])
  File "/home/flip/bin/resSys", line 153, in main
    nb256 = processTemplates(ipfix, 256, directory)
  File "/home/flip/bin/resSys", line 120, in 
    ...
  File "/home/flip/bin/resSys", line 87, in process256
    myDf = (df.groupby(['IP']).agg({ 'Packets': ['sum', 'mean'], 'Timestamp': ['min', 'max', diffMean]}).compute())
  File "/home/flip/.local/lib/python3.7/site-packages/dask/dataframe/groupby.py", line 1847, in agg
    return self.aggregate(arg, split_every=split_every, split_out=split_out)
  File "/home/flip/.local/lib/python3.7/site-packages/dask/dataframe/groupby.py", line 1843, in aggregate
    return super().aggregate(arg, split_every=split_every, split_out=split_out)
  File "/home/flip/.local/lib/python3.7/site-packages/dask/dataframe/groupby.py", line 1623, in aggregate
    sort=self.sort,
  File "/home/flip/.local/lib/python3.7/site-packages/dask/dataframe/core.py", line 5563, in apply_concat_apply
    meta_chunk = _emulate(chunk, *args, udf=True, **chunk_kwargs)
  File "/home/flip/.local/lib/python3.7/site-packages/dask/dataframe/core.py", line 5612, in _emulate
    return func(*_extract_meta(args, True), **_extract_meta(kwargs, True))
  File "/usr/lib/python3.7/contextlib.py", line 130, in __exit__
    self.gen.throw(type, value, traceback)
  File "/home/flip/.local/lib/python3.7/site-packages/dask/dataframe/utils.py", line 197, in raise_on_meta_error
    raise ValueError(msg) from e
ValueError: Metadata inference failed in `_groupby_apply_funcs`.

You have supplied a custom function and Dask is unable to
determine the type of output that that function returns.

To resolve this please provide a meta= keyword.
The docstring of the Dask function you ran should have more information.

Original error is below:
------------------------
NotImplementedError('Can only union MultiIndex with MultiIndex or Index of tuples, try mi.to_flat_index().union(other) instead.')

Traceback:
---------
  ...

问题是dd.Aggregation 不接受“元”参数。

Answer 1

以下是一些可能对您有所帮助的建议。

1.从聚合中删除compute

我认为应该从聚合中删除 compute。据我所知，聚合在 Pandas 数据帧上运行。

diffMean = dd.Aggregation('diffMean', lambda x: x.diff(), lambda x0: x0.mean())    

2.考虑将时间戳转换为int

我尝试通过以下方式重现问题

import dask
import dask.dataframe as dd

# Create some artificial data
df = dask.datasets.timeseries().reset_index()

diffMean = dd.Aggregation('diffMean', lambda x: x.diff(), lambda x0: x0.mean())    
(df.groupby(['name'])
  .agg({ 'x': ['sum', 'mean'], 'timestamp': ['min', 'max', diffMean]})
  .compute())

这失败了，可能是因为它无法处理沿路某处的时间戳列。转换为 int 对我有用：

(df.astype({'timestamp': int})
  .groupby(['name'])
  .agg({ 'x': ['sum', 'mean'], 'timestamp': ['min', 'max', diffMean]})
  .compute())

3.多索引

您的代码引发的原始错误是

NotImplementedError('Can only union MultiIndex with MultiIndex or Index of tuples, try mi.to_flat_index().union(other) instead.')

在您的示例中，您的数据框中似乎没有 MultiIndex，因此我不确定可能存在什么问题。也许这是dask版本的问题？我在2021.06.2 上复制/修复了这个问题