daskでgroupbyしてから行間のdiffをとる

Pandasの場合は以下のようにgroupbyしてからdiffを呼ぶと1行前とのdiffが取れるが、daskにはdiffメソッドがない。

features[num_features + ["customer_ID"]].groupby(["customer_ID"]).diff(1)

daskの場合、map_overlapメソッドを使って実装する。 diffや移動平均のようにwindowをずらしながら計算するような場合はmap_overlapを使うらしい。diffをとる場合のサンプルコードは以下。

    def _get_difference(self, num_features: List[str]) -> DataFrame:
        def diff(df: DataFrame) -> DataFrame:
            return df.diff(1)
        npartitions: int = min(os.cpu_count() * 3, int(len(self._features) / 100))
        ddf: dd.DataFrame = dd.from_pandas(self._features, npartitions=npartitions).set_index("customer_ID")
        ddf = ddf[num_features].map_overlap(diff, 2, 0)
        ddf = ddf.rename(columns={col: col + "_diff_1" for col in num_features})
        ddf = ddf.groupby("customer_ID").last()
        return ddf.reset_index().compute()

参考：

docs.dask.org