Pyspark - 对窗口函数使用两个时间索引答案

【问题标题】：Pyspark - Using two time indices for window functionPyspark - 对窗口函数使用两个时间索引
【发布时间】：2019-09-25 09:14:43
【问题描述】：

我有一个数据框，其中每一行都有两个日期列。我想创建一个窗口函数，其范围在计算特定范围内的行数，其中两个日期列都在该范围内。在下面的情况下，一行的两个时间戳都必须在当前行的时间戳之前，才能包含在计数中。

包含计数列的示例df：

    +---+-----------+-----------+-----+
    | ID|Timestamp_1|Timestamp_2|Count|
    +---+-----------+-----------+-----+
    |  a|          0|          3|    0|
    |  b|          2|          5|    0|
    |  d|          5|          5|    3|
    |  c|          5|          9|    3|
    |  e|          8|         10|    4|
    +---+-----------+-----------+-----+

我尝试创建两个窗口并在这两个窗口上创建新列：

    w_1 = Window.partitionBy().orderBy('Timestamp_1').rangeBetween(Window.unboundedPreceding, 0)
    w_2 = Window.partitionBy().orderBy('Timestamp_2').rangeBetween(Window.unboundedPreceding, 0)

    df = df.withColumn('count', F.count('ID').over(w_1).over(w_2))

但是，这在 Pyspark 中是不允许的，因此会报错。

有什么想法吗？ SQL中的解决方案也很好！

【问题讨论】：

标签： python sql python-3.x pyspark pyspark-sql

【解决方案1】：

自我加入会起作用吗？

from pyspark.sql import functions as F

df_count = (
    df.alias('a')
    .join(
        df.alias('b'),
        (F.col('b.Timestamp_1') <= F.col('a.Timestamp_1')) &
        (F.col('b.Timestamp_2') <= F.col('a.Timestamp_2')),
        'left'
    )
    .groupBy(
        'a.ID'
    )
    .agg(
        F.count('b.ID').alias('count')
    )
)

df = df.join(df_count, 'ID')

【讨论】：