dask groupby результат дублирования первой группы для каждого раздела

Question

Я группирую результаты на основе двух ключей с df, содержащим 10M + строк, 5 столбцов, но после groupby (). Apply (f) для моей собственной определенной функции f, результаты groupby продолжают дублировать первую группу.

original df

Сначала я использую панд, дублирует первую группу.Затем я пробую dask с 4 разделами, он дублирует первую группу этого раздела.

groupby result

apply function

def f(x):
    x['QUANTITY_'] = x['QUANTITY'].shift(len_week_predict).rolling(len_week_train).mean()
    x = x.dropna()
    mae = (x['QUANTITY_'] - x['QUANTITY']).abs().sum()
    mae_f = (x['QUANTITY'].shift(1) - x['QUANTITY']).dropna().abs().sum() * (len(x) / (len(x)-1))
    if mae_f == 0:
        mase = mae / sigma
    else:
        mase = mae / mae_f
    return pd.Series([mae, mae_f, mase], index=['mae', 'mae_f', 'mase'])

Есть ли какие-то проблемы, которые могут возникнуть из-за моей собственной функции f?

import dask.dataframe as dd
ddf_ = dd.from_pandas(df_, npartitions=4)
df_bsl = ddf_.groupby(by=['CUST_NUMBER', 'ITEM_NUM']).apply(f).compute(scheduler='processes')

Я ожидаю, что в группе не будет дублированных результатов.