python cuDF groupby apply with ordered data

Question

我有一些有序的数据，其中有事件的层次结构。每一列都是事件的唯一 ID，相对于层次结构中高于它的事件。类似于每个日的数字在一个月内是唯一的，并且每个月的数字在一年中都是唯一的。我想让最低级别在最高级别中独一无二，就像通过从 1 到 365 的数字使一年中的每一天都独一无二。我的用例不是特定于天、月和年。

以前：

| ID | EVENT_1 | EVENT_2 | EVENT_3 |
| -- | ------- | ------- | ------- |
|  1 |       1 |       1 |       1 |
|  1 |       1 |       1 |       2 |
|  1 |       1 |       1 |       3 |
|  1 |       1 |       2 |       1 |
|  1 |       1 |       2 |       2 |
|  1 |       1 |       3 |       1 |
|  1 |       1 |       3 |       2 |
|  1 |       2 |       1 |       1 |
|  1 |       2 |       1 |       2 |

后：

| ID | EVENT_1 | EVENT_2 | EVENT_3 | EVENT_3A |
| -- | ------- | ------- | ------- | -------- |
|  1 |       1 |       1 |       1 |        1 |
|  1 |       1 |       1 |       2 |        2 |
|  1 |       1 |       1 |       3 |        3 |
|  1 |       1 |       2 |       1 |        4 |
|  1 |       1 |       2 |       2 |        5 |
|  1 |       1 |       3 |       1 |        6 |
|  1 |       1 |       3 |       2 |        7 |
|  1 |       2 |       1 |       1 |        1 |
|  1 |       2 |       1 |       2 |        2 |

目标是获得一列，其中对于每个 id，都有一个EVENT_3A，EVENT_3A是EVENT_3相对于EVENT_1发生的顺序(好像没有EVENT_2一样)。此外，还有许多 ID 必须独立计算。现在我在 CPU 上执行此操作，但这需要很长时间，因此我想切换到在 GPU 上执行此操作。

我的主要想法是做一个groupby('ID').apply_grouped()或groupby('ID').agg()但我不知道在apply_grouped()或agg()函数中放什么。我之前在 CPU 上使用 dask 执行此操作，但它更直观，因为分组的数据帧直接传递给apply()函数。似乎在 cuDF 中，我必须传递 incols，并且我无法弄清楚如何将它们视为数据帧。

大约有 5,000 个 ID，因此理想情况下，每个分组的 ID 将由 GPU 中的一个内核处理，但我不确定它是否可以像那样工作，因为我是 GPU 编程的新手。

任何建议或解决方案都有帮助，谢谢。

Answer 1

目标是获得一列，其中对于每个 id，都有一个EVENT_3A，使得EVENT_3A是EVENT_3相对于EVENT_1发生的顺序(好像没有EVENT_2)。

您描述的是按累积计数操作，键为 [ID， EVENT_1]。它尚未在 cuDF 中实现，因此您可能希望使用用户定义的函数。例如：

您的设置：

import cudf
from numba import cuda
import numpy as np
​
data = {
"ID":[1,1,1,1,1,1,1,1,1],
"EVENT_1":[1,1,1,1,1,1,1,2,2,],
"EVENT_2":[1,1,1,2,2,3,3,1,1],
"EVENT_3":[1,2,3,1,2,1,2,1,2]
}
​
gdf = cudf.DataFrame(data)
print(gdf)
ID  EVENT_1  EVENT_2  EVENT_3
0   1        1        1        1
1   1        1        1        2
2   1        1        1        3
3   1        1        2        1
4   1        1        2        2
5   1        1        3        1
6   1        1        3        2
7   1        2        1        1
8   1        2        1        2

我们可以而且应该在这里使用apply_grouped。我鼓励您查看文档以完全了解此处发生的情况，但在高级别上，我们可以使用该组内线程索引作为该行的索引作为计数。我们传递EVENT_3列，因此确保列名和函数参数匹配。

def cumcount(EVENT_3, cumcount):
for i in range(cuda.threadIdx.x, len(EVENT_3), cuda.blockDim.x):
cumcount[i] = i + 1 # since your exmaple counts start with 1 rather than 0

results = gdf.groupby(["ID", "EVENT_1"]).apply_grouped(cumcount,
incols=['EVENT_3'],
outcols=dict(cumcount=np.int32))
print(results.sort_index()) # get the original row order, for demonstration
ID  EVENT_1  EVENT_2  EVENT_3  cumcount
0   1        1        1        1         1
1   1        1        1        2         2
2   1        1        1        3         3
3   1        1        2        1         4
4   1        1        2        2         5
5   1        1        3        1         6
6   1        1        3        2         7
7   1        2        1        1         1
8   1        2        1        2         2

作为健全性检查，您可以证明这些结果与较大数据上的熊猫匹配。

n_ids = 5000
n_rows = 10000000
​
df = pd.DataFrame({
"ID": np.random.choice(range(n_ids), n_rows),
"EVENT_1": np.random.choice(range(500), n_rows),
"EVENT_2": np.random.choice(range(500), n_rows),
"EVENT_3": np.random.choice(range(n_ids), n_rows)
})
gdf = cudf.from_pandas(df)
results = gdf.groupby(["ID", "EVENT_1"]).apply_grouped(cumcount,
incols=['EVENT_3'],
outcols=dict(cumcount=np.int32))
results = results.sort_index()
pdf_res = df.groupby(["ID", "EVENT_1"]).EVENT_3.cumcount() + 1
print(pdf_res.astype("int32").equals(results['cumcount'].to_pandas()))
True

请注意，如果您有 100 万行和合理数量的组<，则在 pandas 中使用df.groupby([ID, EVENT_1]).EVENT_3.cumcount() + 1可能会很快，因为按 groupby cumcount 相当有效。话虽如此，cuDF UDF在规模上会快得多。