Объединение нескольких групп в pyspark - PullRequest
Новая
       43

Объединение нескольких групп в pyspark

0 голосов
/ 03 мая 2018

Мои данные выглядят так: 

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

Я хочу сгруппировать его по идентификатору (что прекрасно работает!): 

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)))]

А вот где я не знаю, как сделать вложенную группу. Любые советы?

1 Ответ

0 голосов
/ 03 мая 2018

Нет необходимости сериализации до 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]|
#+---+--------+-------+-------+

Обновление

Если вам нужно сохранить порядок действий , лучше всего использовать pyspark.sql.Window с orderBy(). Это потому, что, кажется, существует некоторая двусмысленность относительно того, поддерживает ли groupBy() после orderBy() этот порядок .

Предположим, ваши метки времени хранятся в столбце "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()
...