Python DataFrame - проверяет другие строки на совпадение идентификатора и аналогичной отметки времени

Question

У меня простая проблема с питоном.У меня есть DataFrame журнальных подписок, таких как:

SubId         UserID             Created             Expired
09483          2938              10/9/2018             N/A
03824          3899              10/13/2018            N/A
02853          0838              10/29/2017          10/1/2018 
08992          2938              10/2/2018           10/8/2018

Я хотел бы создать новый логический столбец, который проверяет, была ли у того же UserID предыдущая подписка, которая закончилась незадолго до (<5 дней) этойновая подписка началась </p>

SubId         UserID             Created             Expired        Extension_of_Sub
09483          2938              10/9/2018             N/A            1
03824          3899              10/13/2018            N/A            0
02853          0838              10/29/2017          10/1/2018        0
08992          2938              10/2/2018           10/8/2018        0

Как мне поступить?Другими словами, я пытаюсь получить более точное число «оттока», и переключение с подпункта на один журнал по сравнению с другим, вероятно, не столько отток, сколько миграция.

Спасибо!

Answer 1

Этого можно добиться с помощью объединения и функций pyspark.sql.functions с датой и при .Пожалуйста, найдите прокомментированный пример ниже:

from pyspark.sql import functions as F
from pyspark.sql.types import IntegerType
l = [('09483', '2938', '10/9/2018', None )
,('03824', '3899',  '10/13/2018', None)
,('02853', '0838',   '10/29/2017', '10/1/2018')
,('08992', '2938',    '10/2/2018', '10/8/2018')
]
df = spark.createDataFrame(l,['SubId','UserID','Created','Expired'])
#This cast's the Expired and Created columns to columns of the type date
df = df.withColumn("Expired", F.to_date(df.Expired,  'MM/dd/yyyy'))
df = df.withColumn("Created", F.to_date(df.Created,  'MM/dd/yyyy'))

#Our join will create a dataframe which has two columns with the name UserID. This list will eliminiate the ambiguity
newNames = ['SubId','UserID', 'Created', 'Expired', 'dropUserID', 'dropExpired', 'Extension_of_Sub']

#We create a dateframe with a new column that contains the max Expired date per UserID
tmp = df.select(df.UserID, df.Expired).groupBy(df.UserID).agg(F.max("Expired").alias('maxExpired'))

#This is the join condition, which joins only rows with running subscriptions
cond = [df.UserID == tmp.UserID, df.Expired.isNull()]

#We use a left to join as we also want to keep the expired subscriptions in the dataframe
df = df.join(tmp, cond, 'left') 

#...and finally with can fill the column Extension_of_Sub when the difference between the creation date of a new subscription and the expiry date of an former subscription is less then 5 days. 
df.withColumn('Extension_of_Sub', F.when(F.datediff(df.Created, df.maxExpired) < 5, 1).otherwise(0)).toDF(*newNames).drop('dropUserID', 'dropExpired').show()