从当前时间开始计算N天内的事件

Question

我有一个很简单的问题。我有一个ID和发生时间戳的长列表，我想按ID计算特定时间窗口中的时间戳。以下是示例数据：

+---------------+-------------------+
|             id|         Occurrence|
+---------------+-------------------+
|533ladk203ldpwk|2018-03-28 17:52:04|
|516dlksw9823adp|2018-03-26 12:58:04|
|516dlksw9823adp|2018-01-24 07:52:16|
|533ladk203ldpwk|2018-03-18 03:23:11|
|533ladk203ldpwk|2018-03-14 08:30:13|
+---------------+-------------------+

这就是我想要的回报：

+---------------+-------------------------+
|             id|Last30daysOccurrenceCount|
+---------------+-------------------------+
|533ladk203ldpwk|                        3|
|516dlksw9823adp|                        1|
+---------------+-------------------------+

有没有一种简单的方法可以使用PySpark或SQL来实现这一点？你知道吗

Answer 1

可以使用^{}获取当前时间戳，并使用^{}计算它与"Occurrence"中的值之间的差异。你知道吗

例如：

import pyspark.sql.functions as f
df.withColumn('days_since_today', f.datediff(f.current_timestamp(), f.col("Occurrence")))\
    .show()
#+       -+         -+        +
#|             id|         Occurrence|days_since_today|
#+       -+         -+        +
#|533ladk203ldpwk|2018-03-28 17:52:04|               5|
#|516dlksw9823adp|2018-03-26 12:58:04|               7|
#|516dlksw9823adp|2018-01-24 07:52:16|              68|
#|533ladk203ldpwk|2018-03-18 03:23:11|              15|
#|533ladk203ldpwk|2018-03-14 08:30:13|              19|
#+       -+         -+        +

然后可以筛选出满足"days_since_today"小于或等于30的条件的行，按"id"分组并计数。你知道吗

df.withColumn('days_since_today', f.datediff(f.current_timestamp(), f.col("Occurrence")))\
    .where("days_since_today <= 30")\
    .groupBy('id')\
    .agg(f.count("Occurrence").alias("Last30daysOccurrenceCount"))\
    .show()
#+       -+            -+
#|             id|Last30daysOccurrenceCount|
#+       -+            -+
#|533ladk203ldpwk|                        3|
#|516dlksw9823adp|                        1|
#+       -+            -+

或等效地，不带中间柱：

df.groupBy('id')\
    .agg(
        f.sum(
            f.when(
                f.datediff(f.current_timestamp(), f.col("Occurrence")) <= 30,
                1
            ).otherwise(0)
        ).alias("Last30daysOccurrenceCount")
    )\
    .show()
#+       -+            -+
#|             id|Last30daysOccurrenceCount|
#+       -+            -+
#|533ladk203ldpwk|                        3|
#|516dlksw9823adp|                        1|
#+       -+            -+