Dask `var` и `std` с ddof в контексте группировки и других агрегатах ⇐ Python

[Anonymous]

Предположим, я хочу вычислить дисперсию и/или стандартное отклонение с нестандартным ddof в контексте группировки, я могу это сделать:
df.groupby("a")["b"].var(ddof=2)

Если я хочу, чтобы это происходило вместе с другими агрегатами, я могу использовать:
df.groupby("a").agg(b_var = ("b", "var"), c_sum = ("c", "sum"))

Насколько я понимаю, чтобы иметь возможность использовать ddof не по умолчанию, мне нужно создать собственную агрегацию.
Вот что я получил далеко:
def var(ddof: int = 1) -> dd.Aggregation:
import dask.dataframe as dd

return dd.Aggregation(
name="var",
chunk=lambda s: (s.count(), s.sum(), (s.pow(2)).sum()),
agg=lambda count, sum_, sum_sq: (count.sum(), sum_.sum(), sum_sq.sum()),
finalize=lambda count, sum_, sum_sq: (sum_sq - (sum_ ** 2 / count)) / (count - ddof),
)

Тем не менее, я обнаружил ошибку выполнения:
df.groupby("a").agg({"b": var(2)})

RuntimeError('Не удалось создать метаданные для DecomposableGroupbyAggregation(frame=df, arg={'b': объект
Чего мне не хватает? Есть ли лучший способ добиться этого?
Замена? s.pow(2) с s**2 также приводит к ошибке.
Полный сценарий:
import dask.dataframe as dd

data = {
"a": [1, 1, 1, 1, 2, 2, 2],
"b": range(7),
"c": range(10, 3, -1),
}

df = dd.from_dict(data, 2)

def var(ddof: int = 1) -> dd.Aggregation:
import dask.dataframe as dd

return dd.Aggregation(
name="var",
chunk=lambda s: (s.count(), s.sum(), (s.pow(2)).sum()),
agg=lambda count, sum_, sum_sq: (count.sum(), sum_.sum(), sum_sq.sum()),
finalize=lambda count, sum_, sum_sq: (sum_sq - (sum_ ** 2 / count)) / (count - ddof),
)

df.groupby("a").agg(b_var = ("b", "var"), c_sum = ("c", "sum")) #

Подробнее здесь: https://stackoverflow.com/questions/793 ... gregations