使用自定义聚合函数在Dask中进行GroupBy操作构建模式和相应计数函数

Question

使用自定义聚合函数在Dask中进行GroupBy操作构建模式和相应计数函数

6

所以，dask现在已经更新，支持自定义聚合函数进行分组。（感谢开发团队和@chmp的工作！）我目前正在尝试构建一种模式功能和相应的计数功能。基本上，我设想的是，模式返回一个列表，对于每个分组，它都会返回特定列的最常见值（即[4,1,2]）。此外，还有一个相应的计数函数，返回这些值的实例数量，即3。

现在，我正在尝试在代码中实现这一点。根据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部分应该如何工作方面遇到了困难。对于这个问题的任何帮助都将不胜感激。

谢谢！

- user48944

1个回答

网页内容由stack overflow 提供, 点击上面的

可以查看英文原文，
原文链接

- user8570642 · Accepted Answer

诚然，目前文档的细节有些不足。感谢您引起了我的关注。如果这个答案有帮助，请告诉我，我会为dask贡献更新后的文档。

对于您的问题：对于单个返回值，聚合的不同步骤等效于：

res = chunk(df.groupby('g')['col'])
res = agg(res.groupby(level=[0]))
res = finalize(res)

在这些术语中，模式函数可以如下实现：

def chunk(s):
    # for the comments, assume only a single grouping column, the 
    # implementation can handle multiple group columns.
    #
    # s is a grouped series. value_counts creates a multi-series like 
    # (group, value): count
    return s.value_counts()


def agg(s):
    # s is a grouped multi-index series. In .apply the full sub-df will passed
    # multi-index and all. Group on the value level and sum the counts. The
    # result of the lambda function is a series. Therefore, the result of the 
    # apply is a multi-index series like (group, value): count
    return s.apply(lambda s: s.groupby(level=-1).sum())

    # faster version using pandas internals
    s = s._selected_obj
    return s.groupby(level=list(range(s.index.nlevels))).sum()


def finalize(s):
    # s is a multi-index series of the form (group, value): count. First
    # manually group on the group part of the index. The lambda will receive a
    # sub-series with multi index. Next, drop the group part from the index.
    # Finally, determine the index with the maximum value, i.e., the mode.
    level = list(range(s.index.nlevels - 1))
    return (
        s.groupby(level=level)
        .apply(lambda s: s.reset_index(level=level, drop=True).argmax())
    )

mode = dd.Aggregation('mode', chunk, agg, finalize)

注意，这个实现在出现并列情况时与数据框的.mode函数不匹配。这个版本会返回其中一个值而不是所有值。

现在可以使用众数聚合函数，如下所示：

import pandas as pd
import dask.dataframe as dd

df = pd.DataFrame({
    'col': [0, 1, 1, 2, 3] * 10,
    'g0': [0, 0, 0, 1, 1] * 10,
    'g1': [0, 0, 0, 1, 1] * 10,
})
ddf = dd.from_pandas(df, npartitions=10)

res = ddf.groupby(['g0', 'g1']).agg({'col': mode}).compute()
print(res)