我正在使用Dask处理数据集(考虑到它不适合内存),我想根据列和类型对具有不同聚合函数的实例进行分组。
Dask有一组用于数值数据类型的默认聚合函数,但不适用于字符串/对象。有没有办法为字符串实现用户定义的聚合函数,有点类似于下面的例子?
atts_to_group = {'A', 'B'}
agg_fn = {
'C': 'mean' #int
'D': 'concatenate_fn1' #string - No default fn for strings - Doesn't work
'E': 'concatenate_fn2' #string
}
ddf = ddf.groupby(atts_to_group).agg(agg_fn).compute().reset_index()
此时,我可以在删除不相关的列/行时读取内存中的整个数据集,但我更愿意继续Dask中的处理,因为它可以更快地执行所需的操作。
编辑:尝试直接在字典上添加自定义函数:
def custom_concat(df):
...
return df_concatd
agg_fn = {
'C': 'mean' #int
'D': custom_concat(df)
}
-------------------------------------------------------
ValueError: unknown aggregate Dask DataFrame Structure:
已实现的Dask提供了Aggregation data structure。自定义聚合可以按如下方式完成:
# Concatenates the strings and separates them using ","
custom_concat = dd.Aggregation('custom_sum', lambda x: ",".join(str(x)), lambda x0: ",".join(str(x0)))
custom_concat_E = ...
atts_to_group = {'A', 'B'}
agg_fn = {
'C': 'mean' #int
'D': custom_concat_D
'E': custom_concat_E
}
ddf = ddf.groupby(atts_to_group).agg(agg_fn).compute().reset_index()
这也可以通过Dataframe.apply完成,以获得更简洁的解决方案
def agg_fn(x):
return pd.Series(
dict(
C = x['C'].mean(), # int
D = "{%s}" % ', '.join(x['D']), # string (concat strings)
E = ...
)
)
ddf = ddf.groupby(atts_to_group).apply(agg_fn).compute().reset_index