pandas groupby 对象上的 dask 延迟函数

Question

我不知道如何计算来自 df.groupy.apply() 操作的延迟对象。如果有人可以提供帮助，我真的很感激。这是我写的示例代码

import pandas as pd
import dask
df = pd.DataFrame(columns=['id','id2','val1'])
df['id'] = ['A','A','A','B','C','C','D','D']
df['id2']=['a','a','b','a','a','b','b','b']
df['val1']= [1,2,3,4,5,6,7,8]
@dask.delayed
def dask_test(group,val_col):
    for idx,row in group.iterrows():
        group.loc[idx,'test']=2*group.loc[idx,val_col]
    return group

tmp_grp = df.groupby(['id','id2']).apply(dask_test,'val1')

tmp_grp的输出是

id  id2
A   a      Delayed('copy-f0e26845-fc3a-4bb7-8609-47b923c0...
    b      Delayed('copy-9b6cecf5-9fa4-4301-ba2d-dec5478d...
B   a      Delayed('copy-7b538f4b-ac3f-4c83-b37b-e620d0ba...
C   a      Delayed('copy-c722fa78-c46e-422a-88a5-b9e48cac...
    b      Delayed('copy-01454a03-fd28-4fa5-b487-563ccc66...
D   b      Delayed('copy-f6cf94bd-d457-4495-bb2e-1db0152c...
dtype: object

我不知道如何从中调用延迟对象并计算它们。

在此先感谢您。

Answer 1

使用 delayed 时，最好显式构造延迟值列表，在您的上下文中，这将是：

delayeds=[dask_test(group, 'val1') for _, group in df.groupby(['id', 'id2'])]

然后，可以使用dask.compute(*delayeds)计算延迟值。

Answer 2

由于您使用的是 pandas，您可以考虑使用 Dask DataFrame 而不是 Delayed，这样优化得更好。 :)

类似于：

import dask.dataframe as dd

ddf = dd.from_pandas(df, npartitions=4)

def dask_test(group,val_col):
    for idx,row in group.iterrows():
        group.loc[idx,'test']=2*group.loc[idx,val_col]
    return group

tmp_grp = ddf.groupby(['id','id2']).apply(dask_test,'val1')
tmp_grp.compute().sort_index()

请注意，当您使用 Dask DataFrame 时，您不需要在 dask_test 函数中使用 for 循环。另请参阅：https://docs.dask.org/en/latest/dataframe-groupby.html#difficult-cases 了解优化提示

pandas groupby 对象上的 dask 延迟函数

dask delayed functions on pandas groupby objects

dask

dask-delayed

pandas-groupby