Pyspark - GroupBy и Count в сочетании с ГДЕ

Question

Скажем, у меня есть список подписок на журналы, например:

subscription_id    user_id       created_at
 12384               1           2018-08-10
 83294               1           2018-06-03
 98234               1           2018-04-08
 24903               2           2018-05-08
 32843               2           2018-03-06
 09283               2           2018-04-07

Теперь я хочу добавить столбец, в котором указано, сколько предыдущих подписок было у пользователя до этой текущей подписки.Например, если это первая подписка пользователя, значение нового столбца должно быть 0. Если у них была одна подписка, начинающаяся до этой подписки, значение нового столбца должно быть 1. Вот полный желаемый результат:

subscription_id    user_id       created_at        users_previous_subs
 12384               1           2018-08-10                  2
 83294               1           2018-06-03                  1
 98234               1           2018-04-08                  0
 24903               2           2018-05-08                  2
 32843               2           2018-04-06                  1
 09283               2           2018-03-07                  0

Как мне это сделать, желательно в PySpark, поэтому не используйте shift

Дайте мне знать, если это не ясно.Спасибо !!

Answer 1

Это сводится к row_number вычислению.

from pyspark.sql import Window
from pyspark.sql import functions as func
#Define a window
w = Window.partitionBy(df.user_id).orderBy(df.created_at)
#Add an extra column with rownumber
df.withColumn('prev_subs',func.row_number().over(w)-1)
df.show()

Если могут быть связи (т. Е. Более 1 строки с заданной датой для пользователя), используйте dense_rank.

df.withColumn('prev_subs',func.dense_rank().over(w)-1)