修改后的答案:
您可以在此处使用简单的窗口函数技巧。一堆进口:
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")
Cast date
to 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")
Add 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 开始你应该能够使用a window功能 https://spark.apache.org/docs/latest/api/python/pyspark.sql.html#pyspark.sql.functions.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 |
+---------------------------------------------+-----+
当涉及到时区时你必须要小心一点。