pySpark - 在滚动窗口中获取最大值行

Posted 2023-03-31

【中文标题】pySpark - 在滚动窗口中获取最大值行

【英文标题】：pySpark - get max value row in a rolling window

【发布时间】：2020-04-23 16:32:39

【问题描述】：

我有一个 pyspark 数据框，示例行如下。我试图在 10 分钟内获得最大平均值。我正在尝试使用Window函数，但无法达到结果。

这是我的数据框，其中包含 30 分钟的随机数据。我希望输出 3 行，每 10 分钟 1 行。

+-------------------+---------+
|         event_time|avg_value|
+-------------------+---------+
|2019-12-29 00:01:00|      9.5|
|2019-12-29 00:02:00|      9.0|
|2019-12-29 00:04:00|      8.0|
|2019-12-29 00:06:00|     21.0|
|2019-12-29 00:08:00|      7.0|
|2019-12-29 00:11:00|      8.5|
|2019-12-29 00:12:00|     11.5|
|2019-12-29 00:14:00|      8.0|
|2019-12-29 00:16:00|     31.0|
|2019-12-29 00:18:00|      8.0|
|2019-12-29 00:21:00|      8.0|
|2019-12-29 00:22:00|     16.5|
|2019-12-29 00:24:00|      7.0|
|2019-12-29 00:26:00|     14.0|
|2019-12-29 00:28:00|      7.0|
+-------------------+---------+

我正在使用下面的代码

window_spec = Window.partitionBy('event_time').orderBy('event_time').rangeBetween(-60*10,0)
new_df = data.withColumn('rank', rank().over(window_spec))
new_df.show()

但是这段代码给了我以下错误：

pyspark.sql.utils.AnalysisException: 'Window Frame specifiedwindowframe(RangeFrame, -600, currentrow$()) must match the required frame specifiedwindowframe(RowFrame, unboundedpreceding$(), currentrow$());'

我想要的输出是

+-------------------+---------+
|         event_time|avg_value|
+-------------------+---------+
|2019-12-29 00:06:00|     21.0|
|2019-12-29 00:16:00|     31.0|
|2019-12-29 00:22:00|     16.5|
+-------------------+---------+

有人可以帮我解决这个问题吗？

TIA。

【参考方案1】：

您可以使用 groupBy 和 window。

from pyspark.sql import functions as F
df.groupBy(F.window("event_time","10 minutes"))\
  .agg(F.max("avg_value").alias("avg_value")).show()

#+--------------------+---------+
#|              window|avg_value|
#+--------------------+---------+
#|[2019-12-29 00:20...|     16.5|
#|[2019-12-29 00:10...|     31.0|
#|[2019-12-29 00:00...|     21.0|
#+--------------------+---------+

要获得 event_time 列的确切输出，您可以使用 collect_list、array_sort 和 element_at(spark2.4+)

from pyspark.sql import functions as F
df.groupBy(F.window("event_time","10 minutes"))\
  .agg(F.element_at(F.array_sort(F.collect_list("event_time")),-2).alias("event_time"),\
       F.max("avg_value").alias("avg_value")).drop("window").orderBy("event_time").show()

#+-------------------+---------+
#|event_time         |avg_value|
#+-------------------+---------+
#|2019-12-29 00:06:00|21.0     |
#|2019-12-29 00:16:00|31.0     |
#|2019-12-29 00:26:00|16.5     |
#+-------------------+---------+

UPDATE：

df.groupBy(F.window("event_time","10 minutes"))\
  .agg(F.collect_list(F.struct("event_time","avg_value")).alias("event_time")\
       ,F.max("avg_value").alias("avg_value"))\
  .withColumn("event_time", F.expr("""filter(event_time, x-> x.avg_value=avg_value)"""))\
        .select((F.col("event_time.event_time")[0]).alias("event_time"),"avg_value").orderBy("event_time").show()

#+-------------------+---------+
#|         event_time|avg_value|
#+-------------------+---------+
#|2019-12-29 00:06:00|     21.0|
#|2019-12-29 00:16:00|     31.0|
#|2019-12-29 00:22:00|     16.5|
#+-------------------+---------+

【讨论】：

谢谢。这真的很有帮助。我只能看到最大值与 event_time 不匹配示例：对于 21-30 分钟的窗口，第 22 分钟的最大值为 16.5，而不是第 26 分钟。当我在整个数据帧上运行此代码时，我在几个地方看到了这一点。

完美运行..!!非常感谢。会努力理解你的作品。

np。我使用higher order functionfilter 来遍历两列的结构，然后得到我们需要的时间。看看这些功能，它们很有帮助

【参考方案2】：

您的数据

data = [
    ('2019-12-29 00:01:00', 9.5,),
    ('2019-12-29 00:02:00', 9.0,),
    ('2019-12-29 00:04:00', 8.0,),
    ('2019-12-29 00:06:00', 21.0,),
    ('2019-12-29 00:08:00', 7.0,),
    ('2019-12-29 00:11:00', 8.5,),
    ('2019-12-29 00:12:00', 11.5,),
    ('2019-12-29 00:14:00', 8.0,),
    ('2019-12-29 00:16:00', 31.0,),
    ('2019-12-29 00:18:00', 8.0,),
    ('2019-12-29 00:21:00', 8.0,),
    ('2019-12-29 00:22:00', 16.5,),
    ('2019-12-29 00:24:00', 7.0,),
    ('2019-12-29 00:26:00', 14.0,),
    ('2019-12-29 00:28:00', 7.0,),
]
df = spark.createDataFrame(data, ['event_time', 'avg_value'])

解决方案

from pyspark.sql import Window
from pyspark.sql.functions import window, max, col

w = Window().partitionBy('group_col')

(
    df.
        withColumn(
            'group_col',
            window('event_time', '10 minutes')
        ).
        withColumn(
            'max_val',
            max(col('avg_value')).over(w)
        ).
        where(
            col('avg_value') == col('max_val')
        ).
        drop(
            'max_val',
            'group_col'
        ).
        orderBy('event_time').
        show(truncate=False)
)

+-------------------+---------+                                                 
|event_time         |avg_value|
+-------------------+---------+
|2019-12-29 00:06:00|21.0     |
|2019-12-29 00:16:00|31.0     |
|2019-12-29 00:22:00|16.5     |
+-------------------+---------+