本文介绍一种非递归、高性能的 pyspark 方法，通过窗口函数动态维护历史最大结束时间，实现对按客户与位置分区的时间区间进行连通性分组，准确识别所有重叠行并分配唯一重叠组号。
本文介绍一种非递归、高性能的 pyspark 方法，通过窗口函数动态维护历史最大结束时间，实现对按客户与位置分区的时间区间进行连通性分组，准确识别所有重叠行并分配唯一重叠组号。
在大规模时序数据处理中（如百万级用户行为、服务周期或审批流程记录），常需将存在时间重叠的相邻记录聚合成逻辑“重叠组”——即若当前行的起始时间 ≤ 前序任意行的最大结束时间，则属于同一组；否则开启新组。该问题本质是带状态的流式区间合并（Interval Merging），但传统递归或 UDF 方案在 Spark 上极易引发性能瓶颈甚至 OOM。
关键在于：不能仅比较前一行（lag），而需动态追踪当前行之前所有已归属同一组的记录中最大的 process_end_date。这正是原代码失效的根本原因——它用 lag 获取单一行的前驱结束时间，而实际逻辑要求的是“截至上一行为止，本组已覆盖的最晚截止时间”。
✅ 正确解法：基于 rowsBetween 的累积最大值窗口
我们使用 Window.rowsBetween(Window.unboundedPreceding, Window.currentRow - 1) 精确捕获当前行之前所有行（在相同分区和排序下）的 process_end_date 最大值，再判断当前 process_start_date 是否超出该范围：
from pyspark.sql import SparkSession
from pyspark.sql.functions import col, when, sum as spark_sum, max as spark_max
from pyspark.sql.window import Window
from pyspark.sql.types import DateType
spark = SparkSession.builder.appName("OverlapGrouping").getOrCreate()
# 示例数据（注意：已按 process_start_date 排序）
data = [
(1, 2277953, 'A', '2015-03-13', '2016-04-15'),
(2, 2277953, 'A', '2016-04-04', '2019-12-31'),
(3, 2277953, 'A', '2019-06-06', '2019-06-20'),
(4, 2277953, 'A', '2019-06-30', '2019-12-31'),
(5, 2277953, 'A', '2020-01-01', '2020-12-31'),
(6, 2277953, 'A', '2020-06-30', '2020-12-31')
]
df = spark.createDataFrame(data, ['recordno', 'customerid', 'locationid', 'process_start_date', 'process_end_date'])
df = df.withColumn("process_start_date", col("process_start_date").cast(DateType()))
df = df.withColumn("process_end_date", col("process_end_date").cast(DateType()))
# 定义窗口：按 customerid + locationid 分区，按 process_start_date 升序排列
window_spec = Window.partitionBy("customerid", "locationid").orderBy("process_start_date")
# ✅ 核心步骤：计算当前行之前所有行的最大 process_end_date
df = df.withColumn(
"max_EndDate_prev",
spark_max("process_end_date").over(window_spec.rowsBetween(Window.unboundedPreceding, Window.currentRow - 1))
)
# ✅ 判断是否开启新组：若无前序记录 或 当前行起始 > 前序最大结束时间 → 新组
df = df.withColumn(
"is_new_group",
when(
col("max_EndDate_prev").isNull() | (col("process_start_date") > col("max_EndDate_prev")),
1
).otherwise(0)
)
# ✅ 累加生成组号（等价于“到当前行为止已开启的新组数”）
df = df.withColumn(
"overlap_group",
spark_sum("is_new_group").over(window_spec)
).select(
"recordno", "customerid", "locationid",
"process_start_date", "process_end_date",
"overlap_group"
)
df.show()输出结果：
+--------+----------+----------+------------------+----------------+-------------+ |recordno|customerid|locationid|process_start_date|process_end_date|overlap_group| +--------+----------+----------+------------------+----------------+-------------+ | 1| 2277953| A| 2015-03-13| 2016-04-15| 1| | 2| 2277953| A| 2016-04-04| 2019-12-31| 1| | 3| 2277953| A| 2019-06-06| 2019-06-20| 1| | 4| 2277953| A| 2019-06-30| 2019-12-31| 1| | 5| 2277953| A| 2020-01-01| 2020-12-31| 2| | 6| 2277953| A| 2020-06-30| 2020-12-31| 2| +--------+----------+----------+------------------+----------------+-------------+
⚠️ 注意事项与最佳实践
-
排序前提不可省略：
orderBy("process_start_date")是逻辑正确性的基础。若原始数据未严格按起始时间排序，务必先sort()，否则rowsBetween行为不可预测。 -
空值安全：
isNull()判断必须保留，因首行max_EndDate_prev恒为 null。 -
性能保障：该方案全程使用 Catalyst 优化的内置窗口函数，避免 UDF、
collect()或迭代，可线性扩展至 TB 级数据。 -
扩展性提示：如需进一步输出每组的合并后区间（min start / max end），可在最终结果上按
overlap_group二次聚合。 -
边界验证：当存在完全嵌套（如
[2020-01-01, 2020-12-31]包含[2020-06-30, 2020-12-31]）或紧邻相接（end == next_start）时，当前逻辑默认不合并（因条件为>而非>=）；如需包含相接区间，将>改为>=即可。
此方法以声明式表达替代过程式逻辑，在保证语义精确的同时，充分发挥 Spark 分布式窗口计算优势，是生产环境中处理海量时间重叠分组任务的推荐范式。
