如何使用 Spark 中的组汇总滚动时间 window
How to aggregate over rolling time window with groups in Spark
我有一些数据想按特定列分组，然后根据分组中的滚动时间 window 聚合一系列字段。
下面是一些示例数据：
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 分组，然后创建从最早日期开始的时间 windows 并延长到该组有 30 天没有条目。在这 30 天结束后，下一次 window 将从不属于前 window.
的下一行的日期开始
然后我想汇总，例如获取 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
编辑：抱歉，我意识到我的问题没有正确说明。我实际上想要一个 window 在 30 天不活动后结束。我相应地修改了示例的 group2 部分。
修改后的答案:
您可以在这里使用一个简单的 window 函数技巧。一堆进口货：
from pyspark.sql.functions import coalesce, col, datediff, lag, lit, sum as sum_
from pyspark.sql.window import Window
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 表达式添加到 table:
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。否则，您还必须使用 window 函数。
使用 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 |
+---------------------------------------------+-----+
在时区方面你必须要小心一点。
我有一些数据想按特定列分组，然后根据分组中的滚动时间 window 聚合一系列字段。
下面是一些示例数据：
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 分组，然后创建从最早日期开始的时间 windows 并延长到该组有 30 天没有条目。在这 30 天结束后，下一次 window 将从不属于前 window.
然后我想汇总，例如获取 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
编辑：抱歉，我意识到我的问题没有正确说明。我实际上想要一个 window 在 30 天不活动后结束。我相应地修改了示例的 group2 部分。
修改后的答案:
您可以在这里使用一个简单的 window 函数技巧。一堆进口货：
from pyspark.sql.functions import coalesce, col, datediff, lag, lit, sum as sum_
from pyspark.sql.window import Window
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 表达式添加到 table:
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。否则，您还必须使用 window 函数。
使用 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 |
+---------------------------------------------+-----+
在时区方面你必须要小心一点。