问题描述
我有以下形式的客户数字访问数据框：
|cust_id|datetime|
|1|2020-08-15 15:20|
|1|2020-08-15 16:20|
|1|2020-08-17 12:20|
|1|2020-08-19 14:20|
|1|2020-08-23 09:20|
|2|2020-08-24 08:00|
我想挑选 strong 信号，例如：在5天内至少访问3次的客户。
我最初的想法是，我们必须为每个客户计算所有滑动窗口。
在此示例中，让我们来看cust1：
-
为期5天的窗口，从2020-08-15开始，到2020-08-19结束，总访问量为4
-
从2020-08-16开始至2020-08-20结束的5天窗口，总访问量为2
-
5天的窗口从2020-08-17开始，到2020-08-21结束，总访问量为2
等
所有滑动窗口的最大数量为4。因此cust1符合条件“ 在5天内至少访问了3次”
这似乎是一项昂贵的操作。
您将如何高效地实现这一目标？任何其他想法都欢迎。
解决方法
您可以将datetime列转换为long，并在rangeBetween()函数中传递等于5天的秒数。
from pyspark.sql.functions import *
from pyspark.sql import functions as F
from pyspark.sql.window import Window
df = df.withColumn("date_long",to_date(substring(col("datetime"),10),"yyyy-MM-dd"))\
.withColumn("date_long",unix_timestamp('date_long','yyyy-MM-dd'))
days = lambda i: i * 86400
w = (Window.partitionBy('cust_id').orderBy("date_long").rangeBetween(0,days(5)))
df.withColumn('5_day_visit',F.count("*").over(w)).drop('date_long').show()
+-------+----------------+-----------+
|cust_id| datetime|5_day_visit|
+-------+----------------+-----------+
| 1|2020-08-15 15:20| 4|
| 1|2020-08-15 16:20| 4|
| 1|2020-08-17 12:20| 2|
| 1|2020-08-19 14:20| 2|
| 1|2020-08-23 09:20| 1|
| 2|2020-08-24 08:00| 1|
+-------+----------------+-----------+
要获取每个客户最多5天的访问次数，您可以执行以下操作：
df.withColumn('5_day_visit',F.count("*").over(w)).drop('date_long')\
.groupBy('cust_id').agg(F.max('5_day_visit').alias('max_5_day_visits')).show()
+-------+----------------+
|cust_id|max_5_day_visits|
+-------+----------------+
| 1| 4|
| 2| 1|
+-------+----------------+