我有一些数据要按特定列分组,然后根据组中的滚动时间窗口聚合一系列字段.
以下是一些示例数据：
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 | +---------------------------------------------+-----+
在时区方面你必须要小心一点.
