apache-spark - 如何使用具有时间段约束和其他条件的函数 PySpark 的窗口

Question

我可以获得有关如何在 pyspark 中编写此逻辑的帮助吗？
假设我的表格如图所示。
所以给定:日期，用户ID，访问，分组(旧)作为输入，我想创建一个名为分组(新)的新列，以便它执行以下操作:
对于任何给定的用户 ID:

首先检查并查看分组(旧)是什么。如果是 != Bad，则分组(新)= 分组(旧)

如果分组(旧)=坏，则应用最近日期的最近访问的分组(旧)，使其成为!=坏

但是，如果上一个日期的最新分组(旧)是 != Bad 超过 30 天，则使分组(新)= Bad(因为数据已过期)

我尝试过但没有按预期工作的内容:

days = lambda i: i * 86400 

user_30d_tracker = 
  Window.partitionBy("userid")
    .orderBy(f.col("date").cast("timestamp").cast("long"))
    .rangeBetween(-days(30), 0)
    .rowsBetween(Window.unboundedPreceding, Window.currentRow - 1)


df = (df.withColumn("Grouping(old)_YN",
      f.when(f.col("Grouping(old)")==f.lit("Bad"), "No")
       .otherwise(f.lit("Yes"))))

df = df.withColumn("Grouping_new", 
          f.max(f.when(f.col("Grouping(old)_YN") == f.lit("Yes"), 
          f.col("Grouping(old)"))).over(user_30d_tracker))

Answer 1

假设这是你的数据框的样子

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

data = [[123, "20200101", 1, "Good_sub1"],
        [123, "20200101", 2, "Bad"],
        [123, "20200115", 1, "Bad"],
        [123, "20200115", 2, "Bad"],
        [123, "20200116", 1, "Good_sub2"],
        [123, "20200116", 2, "Bad"],
        [123, "20200116", 3, "Good_sub3"],
        [123, "20220901", 1, "Bad"]]

df = spark.createDataFrame(data, 
                           "userid:int, date:string, visit:int, `grouping(old)`:string")
df.show()
# +------+--------+-----+-------------+
# |userid|    date|visit|grouping(old)|
# +------+--------+-----+-------------+
# |   123|20200101|    1|    Good_sub1|
# |   123|20200101|    2|          Bad|
# |   123|20200115|    1|          Bad|
# |   123|20200115|    2|          Bad|
# |   123|20200116|    1|    Good_sub2|
# |   123|20200116|    2|          Bad|
# |   123|20200116|    3|    Good_sub3|
# |   123|20220901|    1|          Bad|
# +------+--------+-----+-------------+

days = lambda i: i * 86400 

user_30d_tracker = Window.partitionBy("userid")\
                         .orderBy(f.col("date").cast("timestamp").cast("long"))\
                         .rangeBetween(-days(30), 0)\
                         .rowsBetween(Window.unboundedPreceding, Window.currentRow - 1)

我们来看看 user_30d_tracker window 。考虑到以下几点，此窗口需要进行一些更改

窗口的顺序似乎需要visit的值date的unix时间戳旁边的列.只有 date 的时间戳在 orderBy子句，例如，spark 不能保证访问 = 1 的行会在访问 = 2 之前出现。所以，我们需要以某种方式包含 visit orderBy中的栏目条款。

附加框架rowsBetween将覆盖 rangeBetween框架，因此它将与预期的不同。

一个选项将只使用 rangeBetween window 的框架。但是，由于 rangeBetween框架只允许在 orderBy 中使用 1 列表达式，我们可以通过添加 visit 来使用解决方法值为 date 的 unix 时间戳(这就像将 visit 视为自 date 开始以来已经过去了多少秒)。

user_30d_tracker =  Window\
                        .partitionBy("userid")\
                        .orderBy(f.unix_timestamp("date", "yyyyMMdd") + f.col("visit"))\
                        .rangeBetween(-days(30), 0)

然后，要获得最新的非“坏”分组(旧)值，最好使用 last函数(使用 ignorenulls=True )代替 max因为它采用窗口中的最近值，而不是排序字符串的最大值。之后，使用 coalesce填充新列中的空值。

df = (df
      .withColumn("Grouping(old)_YN",
                  f.when(f.col("Grouping(old)") == f.lit("Bad"), "No")
                  .otherwise(f.lit("Yes")))
      .withColumn("Grouping_new", 
                  f.last(f.when(f.col("Grouping(old)_YN") == f.lit("Yes"), 
                                f.col("Grouping(old)")), ignorenulls=True).over(user_30d_tracker))
      .withColumn("Grouping_new", f.coalesce(f.col("Grouping_new"), f.col("Grouping(old)")))
      )
df.show()

# +------+--------+-----+-------------+----------------+------------+
# |userid|    date|visit|grouping(old)|Grouping(old)_YN|Grouping_new|
# +------+--------+-----+-------------+----------------+------------+
# |   123|20200101|    1|    Good_sub1|             Yes|   Good_sub1|
# |   123|20200101|    2|          Bad|              No|   Good_sub1|
# |   123|20200115|    1|          Bad|              No|   Good_sub1|
# |   123|20200115|    2|          Bad|              No|   Good_sub1|
# |   123|20200116|    1|    Good_sub2|             Yes|   Good_sub2|
# |   123|20200116|    2|          Bad|              No|   Good_sub2|
# |   123|20200116|    3|    Good_sub3|             Yes|   Good_sub3|
# |   123|20220901|    1|          Bad|              No|         Bad|
# +------+--------+-----+-------------+----------------+------------+