时间戳

首页 > 解决方案 > Pyspark 计数任何 ID 的任何滑动窗口

python - Pyspark 计数任何 ID 的任何滑动窗口

问题描述

随着时间的推移，我有一个客户数字访问的数据框，格式如下：

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

我想挑选出强烈的信号，例如：在 5 天内至少访问 3 次的客户。

我最初的想法是我们必须为每个客户计算所有滑动窗口。

在此示例中，我们以 cust1 为例：

等等

所有滑动窗口的最大计数为 4。因此 cust1 符合条件“在 5 天内至少访问过 3 次”

这似乎是一项昂贵的操作。

您将如何有效地实现这一点？欢迎任何其他想法。

标签： pythonapache-sparkpysparkwindow-functionssliding-window

解决方案

您可以在rangeBetween()函数中将该datetime列转换为long并传入相当于 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"),0,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|
+-------+----------------+

推荐阅读