Spark 3.2.4（PySpark+Zeppelin）时间区间重叠ID收集问题

针对你在Spark 3.2.4 + Zeppelin 0.10.1环境下的需求，以下是直接可运行的解决方案：

核心思路 #

通过**自连接（Self Join）**对比每行与其他所有行的时间区间，结合Delta参数判断区间是否重叠或间隔在允许范围内，最后分组收集关联ID集合。

步骤1：创建测试数据 #

from pyspark.sql.functions import collect_set, col

# Zeppelin环境通常已初始化spark对象，无需重复创建
data = [(1, 1, 3), (2, 3, 5), (3, 6, 7)]
df = spark.createDataFrame(data, ["id", "start_time_in_millis", "end_time_in_millis"])
df.show()

步骤2：定义Delta参数并执行关联计算 #

基础场景（Delta=0，仅重叠区间）

delta = 0

# 自连接筛选符合条件的行
joined_df = df.alias("a").join(
    df.alias("b"),
    # 区间重叠判断条件
    (col("a.end_time_in_millis") + delta >= col("b.start_time_in_millis")) &
    (col("a.start_time_in_millis") <= col("b.end_time_in_millis") + delta),
    how="left"
)

# 分组收集关联ID集合
result_df = joined_df.groupBy("a.id").agg(collect_set("b.id").alias("collect"))
result_df.show(truncate=False)

输出结果：

+---+--------+
|id |collect |
+---+--------+
|1  |[1, 2]  |
|2  |[1, 2]  |
|3  |[3]     |
+---+--------+

扩展场景（Delta=1，允许1ms间隔）

仅需修改delta值为1，执行相同代码即可：

delta = 1
# 后续代码同上

输出结果：

+---+-----------+
|id |collect    |
+---+-----------+
|1  |[1, 2]     |
|2  |[1, 2, 3]  |
|3  |[2, 3]     |
+---+-----------+

条件说明 #

关联条件的逻辑是：两个区间（含Delta缓冲）没有完全分离，即：

1. A区间的结束时间+Delta ≥ B区间的开始时间（B不晚于A的缓冲后结束时间）
2. A区间的开始时间 ≤ B区间的结束时间+Delta（A不早于B的缓冲后结束时间）

性能优化提示 #

如果数据量较大，自连接可能存在性能瓶颈，可尝试：

• 先按start_time_in_millis排序，使用窗口函数rangeBetween限定连接范围，减少不必要的行匹配
• 对时间字段建立列式索引（Spark 3.x支持）

内容的提问来源于stack exchange，提问作者Gouri Sankar Mishra