SQL/PySpark：创建一个包含过去 n 天的行数的新列

Question

目前，我有一个由encounter_id 和date 字段组成的表，如下所示：

+---------------------------+--------------------------+
|encounter_id               |date                      |
+---------------------------+--------------------------+
|random_id34234             |2018-09-17 21:53:08.999999|
|this_can_be_anything2432432|2018-09-18 18:37:57.000000|
|423432                     |2018-09-11 21:00:36.000000|
+---------------------------+--------------------------+

encounter_id 是一个随机字符串。

我的目标是创建一个包含过去 30 天内遭遇总数的列。

+---------------------------+--------------------------+---------------------------+
|encounter_id               |date                      | encounters_in_past_30_days|
+---------------------------+--------------------------+---------------------------+
|random_id34234             |2018-09-17 21:53:08.999999| 2                         |
|this_can_be_anything2432432|2018-09-18 18:37:57.000000| 3                         |
|423432                     |2018-09-11 21:00:36.000000| 1                         |
+---------------------------+--------------------------+---------------------------+

目前，我正在考虑以某种方式使用窗口函数并指定一个聚合函数。

感谢您的时间。

Answer 1

这是一种可能的解决方案，我添加了一些示例数据。正如您自己建议的那样，它确实使用了窗口函数。希望这会有所帮助！

import pyspark.sql.functions as F
from pyspark.sql.window import Window

df = sqlContext.createDataFrame(
    [
     ('A','2018-10-01 00:15:00'),
     ('B','2018-10-11 00:30:00'),
     ('C','2018-10-21 00:45:00'),
     ('D','2018-11-10 00:00:00'),
     ('E','2018-12-20 00:15:00'),
     ('F','2018-12-30 00:30:00')
    ],
    ("encounter_id","date")
)

df = df.withColumn('timestamp',F.col('date').astype('Timestamp').cast("long"))
w = Window.orderBy('timestamp').rangeBetween(-60*60*24*30,0)
df = df.withColumn('encounters_past_30_days',F.count('encounter_id').over(w))
df.show()

输出：

+------------+-------------------+----------+-----------------------+
|encounter_id|               date| timestamp|encounters_past_30_days|
+------------+-------------------+----------+-----------------------+
|           A|2018-10-01 00:15:00|1538345700|                      1|
|           B|2018-10-11 00:30:00|1539210600|                      2|
|           C|2018-10-21 00:45:00|1540075500|                      3|
|           D|2018-11-10 00:00:00|1541804400|                      2|
|           E|2018-12-20 00:15:00|1545261300|                      1|
|           F|2018-12-30 00:30:00|1546126200|                      2|
+------------+-------------------+----------+-----------------------+

---

编辑：如果您想以天为粒度，您可以首先将您的日期列转换为 Date 类型。下面的示例，假设五天的窗口意味着今天和之前的四天。如果应该是今天和过去五天，只需删除 -1。

import pyspark.sql.functions as F
from pyspark.sql.window import Window

n_days = 5

df = sqlContext.createDataFrame(
    [
     ('A','2018-10-01 23:15:00'),
     ('B','2018-10-02 00:30:00'),
     ('C','2018-10-05 05:45:00'),
     ('D','2018-10-06 00:15:00'),
     ('E','2018-10-07 00:15:00'),
     ('F','2018-10-10 21:30:00')
    ],
    ("encounter_id","date")
)

df = df.withColumn('timestamp',F.to_date(F.col('date')).astype('Timestamp').cast("long"))
w = Window.orderBy('timestamp').rangeBetween(-60*60*24*(n_days-1),0)
df = df.withColumn('encounters_past_n_days',F.count('encounter_id').over(w))
df.show()

输出：

+------------+-------------------+----------+----------------------+
|encounter_id|               date| timestamp|encounters_past_n_days|
+------------+-------------------+----------+----------------------+
|           A|2018-10-01 23:15:00|1538344800|                     1|
|           B|2018-10-02 00:30:00|1538431200|                     2|
|           C|2018-10-05 05:45:00|1538690400|                     3|
|           D|2018-10-06 00:15:00|1538776800|                     3|
|           E|2018-10-07 00:15:00|1538863200|                     3|
|           F|2018-10-10 21:30:00|1539122400|                     3|
+------------+-------------------+----------+----------------------+