在pyspark中链接多个groupBy

Question

我的数据如下所示：

   id | duration | action1 | action2 | ...
 ---------------------------------------------
    1 | 10       |   A     |   D
    1 | 10       |   B     |   E 
    2 | 25       |   A     |   E
    1 | 7        |   A     |   G

我想按 ID 对其进行分组（效果很好！）：

df.rdd.groupBy(lambda x: x['id']).mapValues(list).collect()

现在我想按持续时间对每个组中的值进行分组，以获得如下结果：

    [(id=1,
      ((duration=10,[(action1=A,action2=D),(action1=B,action2=E),
       (duration=7,(action1=A,action2=G)),

     (id=2,
       ((duration=25,(action1=A,action2=E)))]

这里是我不知道如何进行嵌套分组的地方。有什么建议吗？

Answer 1

无需序列化为rdd。这是一种按多列分组并将其余列聚合到列表中的通用方法，而无需对所有列进行硬编码：

from pyspark.sql.functions import collect_list
grouping_cols = ["id", "duration"]
other_cols = [c for c in df.columns if c not in grouping_cols]
df.groupBy(grouping_cols).agg(*[collect_list(c).alias(c) for c in other_cols]).show()
#+---+--------+-------+-------+
#| id|duration|action1|action2|
#+---+--------+-------+-------+
#|  1|      10| [A, B]| [D, E]|
#|  2|      25|    [A]|    [E]|
#|  1|       7|    [A]|    [G]|
#+---+--------+-------+-------+

更新

如果您需要preserve the order 的操作，最好的方法是使用pyspark.sql.Window 和orderBy()。这是因为对于 groupBy() 是否跟在 orderBy() maintains that order 之后似乎存在一些歧义。

假设您的时间戳存储在"ts" 列中。您应该能够执行以下操作：

from pyspark.sql import Window
w = Window.partitionBy(grouping_cols).orderBy("ts")
grouped_df = df.select(
    *(grouping_cols + [collect_list(c).over(w).alias(c) for c in other_cols])
).distinct()