python - 如何将自定义函数应用于 dask 数据框中的组，使用多列作为函数输入

Question

我有一个非常大的数据框，我正在使用 dask 处理它。数据框大体上看起来像这样:

Col_1    Col_2   Bool_1   Bool_2
A        1       True     False
B        1       True     True
C        1       False    False
D        1       True     False
A        2       False    True
B        2       False    False
C        2       True     False
D        2       True     True

但它有数百万行。

我在这段代码中尝试做的是为每个组计算 Bool_1 和 Bool_2 之间的Jaccard 距离在 Col_2 中形成。这是因为该程序的目的是为 Col_2 中存在的每个组生成一行(每行都有多个统计信息，我只报告相关列).

为此，我首先使用 df.groupby("Col_2") 按 Col_2 对数据帧进行分组，但随后我不知道如何进行。到目前为止，我尝试的每一次尝试都引发了错误。

1:我尝试定义一个函数 compute_jacc_dist() 并通过 apply(compute_jacc_dist, axis=1) 将其传递给组，但它有 args 和 kwargs 的问题(尤其是轴，请参阅 https://github.com/dask/dask/issues/1572 ，我还无法解决)。

2:我尝试使用 from dask_distance import jaccard 并用它来计算 Bool_1 和 Bool_2< 之间的 J 距离 但它会产生奇怪的结果(即使没有交集，每个组也会返回 J=1)。

3:我尝试compute() 数据框并使用以下方法迭代组:

for name, group in df.groupby("Col_2"):
   jacc = dask_distance.jaccard(group["Bool_1"], group["Bool_2"])

但是这个非常慢，因为它会触发计算，然后逐组对如此庞大的数据帧进行操作(即我不想使用它)。作为引用，具有此功能的脚本已运行两天，而我估计我尝试过的任何解决方案 #1 和 #2 如果设置正确，将在 1-2 小时内返回结果。

关于如何处理这个问题有什么建议吗？我理想的解决方案是以正确的方式使用 df.groupby("Col_1").apply(compute_jacc_dist)。非常感谢任何帮助!

Answer 1

经过几个小时的尝试，我是这样做的。如果您正在阅读本文，您可能想阅读本文 ( How to apply euclidean distance function to a groupby object in pandas dataframe? ) 和本文 ( Apply multiple functions to multiple groupby columns )。

def my_function(x):

    d = {}
    v1 = np.array(x["Bool_1"])
    v2 = np.array(x["Bool_2"])
    intersection = np.logical_and(v1, v2).sum()
    union = np.logical_or(v1, v2).sum()
    d["Jaccard"] = float(intersection) / float(union)
    return pd.Series(d, index=["Jaccard"])

df = df.groupby("Col_2").apply(my_function, meta={"Jaccard":"float16"}).compute()

解释

我创建了一个函数来计算数据帧两列之间的 Jaccard 距离。在该函数中，我创建了一个字典 (d)，它将包含我的计算结果。

拥有字典的好处是我可以添加任意多的计算，尽管这里只有一个。

然后该函数返回一个包含字典的 pd.Series。

该函数应用于基于 Col_2 的数据帧组。 meta 数据类型在 apply() 中指定，并且整个事情在最后有 compute()，因为它是一个 dask 数据框和一个必须触发计算才能得到结果。

apply() 应该有与输出列一样多的 meta。