我有一些数据要按特定列分组,然后根据组中的滚动时间窗口聚合一系列字段.
以下是一些示例数据:
df = spark.createDataFrame([Row(date='2016-01-01', group_by='group1', get_avg=5, get_first=1),
Row(date='2016-01-10', group_by='group1', get_avg=5, get_first=2),
Row(date='2016-02-01', group_by='group2', get_avg=10, get_first=3),
Row(date='2016-02-28', group_by='group2', get_avg=20, get_first=3),
Row(date='2016-02-29', group_by='group2', get_avg=30, get_first=3),
Row(date='2016-04-02', group_by='group2', 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")
演员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")
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|
+--------+--------+------------+
first聚合没有意义,但如果列单调增加,您可以使用min.否则你也必须使用窗口函数.
使用Spark 2.1进行测试.Window与早期的Spark版本一起使用时,可能需要子查询和实例.
原始答案(与指定范围无关)
从Spark 2.0开始,你应该可以使用一个window函数:
在给定时间戳指定列的情况下将行划分为一个或多个时间窗口.窗口开始是包含的,但窗口结束是独占的,例如12:05将在窗口[12:05,12:10]但不在[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 |
+---------------------------------------------+-----+
在时区方面你必须要小心一点.
京公网安备 11010802040832号 | 京ICP备19059560号-6 