所以,dask现在已经更新,支持自定义聚合函数进行分组。 (感谢开发团队和@chmp的工作!)我目前正在尝试构建一种模式功能和相应的计数功能。基本上,我设想的是,模式返回一个列表,对于每个分组,它都会返回特定列的最常见值(即[4,1,2])。此外,还有一个相应的计数函数,返回这些值的实例数量,即3。
现在,我正在尝试在代码中实现这一点。根据groupby.py文件,自定义聚合的参数如下:
现在,我正在尝试在代码中实现这一点。根据groupby.py文件,自定义聚合的参数如下:
Parameters
----------
name : str
the name of the aggregation. It should be unique, since intermediate
result will be identified by this name.
chunk : callable
a function that will be called with the grouped column of each
partition. It can either return a single series or a tuple of series.
The index has to be equal to the groups.
agg : callable
a function that will be called to aggregate the results of each chunk.
Again the argument(s) will be grouped series. If ``chunk`` returned a
tuple, ``agg`` will be called with all of them as individual positional
arguments.
finalize : callable
an optional finalizer that will be called with the results from the
aggregation.
这里提供的是计算平均值的代码:
custom_mean = dd.Aggregation(
'custom_mean',
lambda s: (s.count(), s.sum()),
lambda count, sum: (count.sum(), sum.sum()),
lambda count, sum: sum / count,
)
df.groupby('g').agg(custom_mean)
我正在尝试思考最佳实现方式。目前我有以下函数:
def custom_count(x):
count = Counter(x)
freq_list = count.values()
max_cnt = max(freq_list)
total = freq_list.count(max_cnt)
return count.most_common(total)
custom_mode = dd.Aggregation(
'custom_mode',
lambda s: custom_count(s),
lambda s1: s1.extend(),
lambda s2: ......
)
然而我在理解agg部分应该如何工作方面遇到了困难。对于这个问题的任何帮助都将不胜感激。
谢谢!
_agg_finalize
中失败。Traceback是: Traceback(最近的调用最先): File“C:\ ProgramData \ Anaconda2 \ lib \ site-packages \ dask-0.15.2 + 14.g8d906032-py2.7.egg \ dask \ dataframe \ groupby.py”,第1130行,在agg中 返回self.aggregate(arg,split_every = split_every,split_out = split_out) ...... ValueError:元数据推断在_agg_finalize
中失败。我会尝试获取一个MCV示例。 - user48944