将不同事件的计数结果拆分到 pyspark 中的不同列中

Question

我有一个 rdd，我需要从中提取多个事件的计数。最初的 rdd 看起来像这样

+----------+--------------------+-------------------+
|     event|                user|                day|
+----------+--------------------+-------------------+
|event_x   |user_A              |                  0|
|event_y   |user_A              |                  2|
|event_x   |user_B              |                  2|
|event_y   |user_B              |                  1|
|event_x   |user_A              |                  0|
|event_x   |user_B              |                  1|
|event_y   |user_B              |                  2|
|event_y   |user_A              |                  1|
+----------+--------------------+-------------------+

我需要为每种类型的事件（在本例中为 2 种类型的事件：event_x 和 event_y）提供一个计数列，按玩家和日期分组。到目前为止，我只用了一个事件就做到了，结果如下：

+--------------------+-------------------+------------+
|                user|                day|count(event)|
+--------------------+-------------------+------------+
|user_A              |                  0|          11|
|user_A              |                  1|           8|
|user_A              |                  2|           4|
|user_B              |                  0|           2|
|user_B              |                  1|           1|
|user_B              |                  2|          25|
+--------------------+-------------------+------------+

但是我需要任意多列，列数与上面显示的第一个 rdd 的最左边列中出现的事件数相同。所以，如果我只有 2 个事件（x 和 y），它应该是这样的：

+--------------------+-------------------+--------------+--------------+
|                user|                day|count(event_x)|count(event_y)|
+--------------------+-------------------+--------------+--------------+
|user_A              |                  0|            11|             3|
|user_A              |                  1|             8|            23| 
|user_A              |                  2|             4|             2|
|user_B              |                  0|             2|             0|
|user_B              |                  1|             1|             1|
|user_B              |                  2|            25|            11|
+--------------------+-------------------+--------------+--------------+

我目前的密码是：

rdd = rdd.groupby('user', 'day').agg({'event': 'count'}).orderBy('user', 'day')

我应该怎么做才能达到预期的效果？

提前致谢;)

Answer 1

您可以尝试使用数据透视选项进行分组

df =spark.createDataFrame([["event_x","user_A",0],["event_y","user_A",2],["event_x","user_B",2],["event_y","user_B",1],["event_x","user_A",0],["event_x","user_B",1],["event_y","user_B",2],["event_y","user_A",1]],["event","user","day"])

>>> df.show()
+-------+------+---+                                                            
|  event|  user|day|
+-------+------+---+
|event_x|user_A|  0|
|event_y|user_A|  2|
|event_x|user_B|  2|
|event_y|user_B|  1|
|event_x|user_A|  0|
|event_x|user_B|  1|
|event_y|user_B|  2|
|event_y|user_A|  1|
+-------+------+---+

>>> df.groupBy(["user","day"]).pivot("event").agg({"event":"count"}).show()
+------+---+-------+-------+
|  user|day|event_x|event_y|
+------+---+-------+-------+
|user_A|  0|      2|   null|
|user_B|  1|      1|      1|
|user_A|  2|   null|      1|
|user_A|  1|   null|      1|
|user_B|  2|      1|      1|
+------+---+-------+-------+

请看一看，如果您对此有任何疑问，请告诉我。