在pyspark中计算窗口上列的百分位数

Posted 2023-04-15

【中文标题】在pyspark中计算窗口上列的百分位数

【英文标题】：calculate percentile of column over window in pyspark

【发布时间】：2020-06-30 06:06:01

【问题描述】：

我有一个用例，我需要在滑动窗口上计算列（我们称之为 X）的百分位数。 所以窗口定义是按时间顺序排列的 - 过去 120 天：

days = lambda i: i * 86400
w = Window.partitionBy("entityId").orderBy(F.col("trn_time").cast("long").asc())
    .rangeBetween(-days(120),-days(1))

我想过使用 approxQuantile 但它是一个 Dataframe 函数。 第二个选项是使用：

percent_rank().over(w)

但我需要按要在其上进行百分位数的数字列 (X) 对窗口进行排序，并且窗口已经按时间排序。 当我尝试在窗口定义中将 X 添加到 orderBY 时：

w = Window.partitionBy("entityId").orderBy(F.col("trn_time").cast("long").asc(),"X")\
    .rangeBetween(-days(120),-days(1))

我收到以下错误：“不能在具有多个 order by 表达式的窗口规范中使用具有值边界的范围窗口框架”

我该如何实现这个逻辑？

【问题讨论】：

您可以尝试使用 row_number() 的窗口函数按时间和 X 对 df 进行排序，然后在最终计算中按该行号而不是时间和 X 进行排序

【参考方案1】：

你需要把它写成一个内置的 SQL 表达式：

# This is like a UDF
magic_percentile = F.expr('percentile_approx(X, 0.5)')
# Define your window
w = Window.partitionBy("entityId").orderBy(F.col("trn_time").cast("long").asc())
    .rangeBetween(-days(120),-days(1))

df = df.withColumn("rolling_percentile", magic_percentile.over(w))

在计算百分位数时，您总是将值从小到大排序，然后取分位数，这样窗口内的值就会被排序。

参考：Median / quantiles within PySpark groupBy

【讨论】：

感谢@pissall，但是百分位数何时对值进行排序，这是缺少的部分，在您共享的示例中，只需在 X 上执行 percentile_approx ，而不对 X 进行排序。窗口按顺序排序时间。

@user1450410 排序会发生在窗口函数内部，你是看不到的。