sql – 如何在滚动时间窗口中与Spark中的组进行聚合

以下是一些示例数据：

df = spark.createDataFrame([Row(date='2016-01-01',group_by='group1',get_avg=5,get_first=1),Row(date='2016-01-10',get_first=2),Row(date='2016-02-01',group_by='group2',get_avg=10,get_first=3),Row(date='2016-02-28',get_avg=20,Row(date='2016-02-29',get_avg=30,Row(date='2016-04-02',get_avg=8,get_first=4)])

我想按group_by进行分组,然后创建时间窗口,这些时间窗口从最早的日期开始并延长到30天没有该组的条目.在这30天结束后,下一个时间窗口将从下一行的日期开始,该日期不会落在上一个窗口中.

然后我想聚合,例如获取get_avg的平均值,以及get_first的第一个结果.

所以这个例子的输出应该是：

group_by    first date of window    get_avg  get_first
group1      2016-01-01              5        1
group2      2016-02-01              20       3
group2      2016-04-02              8        4

编辑：对不起我意识到我的问题没有正确指定.我实际上想要一个在30天不活动后结束的窗口.我相应地修改了示例的group2部分.

解决方法

你可以在这里使用一个简单的窗口函数技巧.一堆进口：

from pyspark.sql.functions import coalesce,col,datediff,lag,lit,sum as sum_
from pyspark.sql.window import Window

窗口定义：

w = Window.partitionBy("group_by").orderBy("date")

投射日期为DateType：

df_ = df.withColumn("date",col("date").cast("date"))

定义以下表达式：

# Difference from the previous record or 0 if this is the first one
diff = coalesce(datediff("date",lag("date",1).over(w)),lit(0))

# 0 if diff <= 30,1 otherwise
indicator = (diff > 30).cast("integer")

# Cumulative sum of indicators over the window
subgroup = sum_(indicator).over(w).alias("subgroup")

将子组表达式添加到表中：

df_.select("*",subgroup).groupBy("group_by","subgroup").avg("get_avg")

+--------+--------+------------+
|group_by|subgroup|avg(get_avg)|
+--------+--------+------------+
|  group1|       0|         5.0|
|  group2|       0|        20.0|
|  group2|       1|         8.0|
+--------+--------+------------+

第一个对聚合没有意义,但如果列单调增加,则可以使用min.否则你也必须使用窗口函数.

使用Spark 2.1进行测试.与早期的Spark版本一起使用时,可能需要子查询和Window实例.

原始答案(与指定范围无关)

从Spark 2.0开始,您应该可以使用a window function：

Bucketize rows into one or more time windows given a timestamp specifying column. Window starts are inclusive but the window ends are exclusive,e.g. 12:05 will be in the window [12:05,12:10) but not in [12:00,12:05).

from pyspark.sql.functions import window

df.groupBy(window("date",windowDuration="30 days")).count()

但你可以从结果中看到,

+---------------------------------------------+-----+
|window                                       |count|
+---------------------------------------------+-----+
|[2016-01-30 01:00:00.0,2016-02-29 01:00:00.0]|1    |
|[2015-12-31 01:00:00.0,2016-01-30 01:00:00.0]|2    |
|[2016-03-30 02:00:00.0,2016-04-29 02:00:00.0]|1    |
+---------------------------------------------+-----+

在时区方面你必须要小心一点.

（编辑：李大同）

【声明】本站内容均来自网络，其相关言论仅代表作者个人观点，不代表本站立场。若无意侵犯到您的权利，请及时与联系站长删除相关内容!