PySpark中如何找出值低于阈值时段的最小/最大时间戳

现有如下结构的PySpark DataFrame：

需要找出value低于5的连续时段的最小时间（min_time）和最大时间（max_time），期望输出如下：

目前已筛选出value<5的数据并计算了全局的最小和最大时间，但无法按连续时段分组得到预期结果，现有代码如下：

df1=df.filter(col('value')<5)
df1=(df1.withColumn('min_time',min(col('time_in_seconds'))
        .withColumn('max_time',max(col('time_in_seconds')))
df=df.join(df1,['time_in_seconds','value'],'left')

请问如何实现按连续时段分组计算对应min_time和max_time？

要实现连续时段的分组，核心是识别出value<5的连续时间区间，通过生成分组标记ID来区分不同连续区间，具体步骤如下：

1. 标记符合条件的行：新增列标记哪些行满足value<5
2. 生成连续分组ID：结合时间列的连续性与标记列，为每个连续区间生成唯一ID
3. 计算分组的min/max时间：按分组ID聚合得到对应区间的最小和最大时间
4. 关联回原表：将计算结果关联到原DataFrame，非符合条件的行自动填充Null

完整代码实现 #

from pyspark.sql import SparkSession
from pyspark.sql import functions as F
from pyspark.sql.window import Window

# 初始化SparkSession（如果未初始化）
spark = SparkSession.builder.appName("continuous_interval").getOrCreate()

# 构建示例DataFrame
data = [
    (1, 4.5), (2, 4), (3, 3), (4, 5), (5, 6), (6, 7),
    (7, 6), (8, 5), (9, 4.5), (10, 4.2), (11, 3), (12, 3.5)
]
df = spark.createDataFrame(data, ["time_is_seconds", "value"])

# 步骤1：标记符合条件的行
df = df.withColumn("is_below_5", F.when(F.col("value") < 5, 1).otherwise(0))

# 步骤2：生成连续分组ID
window_spec = Window.orderBy("time_is_seconds")
df = df.withColumn(
    "group_id",
    F.sum(
        F.when(
            (F.col("is_below_5") == 1) & 
            (F.lag("is_below_5").over(window_spec) != 1),
            1
        ).otherwise(0)
    ).over(window_spec)
)
# 非符合条件的行group_id设为Null
df = df.withColumn("group_id", F.when(F.col("is_below_5") == 1, F.col("group_id")).otherwise(None))

# 步骤3：计算每个分组的min_time和max_time
grouped_df = df.filter(F.col("group_id").isNotNull()) \
               .groupBy("group_id") \
               .agg(
                   F.min("time_is_seconds").alias("min_time"),
                   F.max("time_is_seconds").alias("max_time")
               )

# 步骤4：关联回原表
result_df = df.join(grouped_df, on="group_id", how="left") \
              .select("time_is_seconds", "value", "min_time", "max_time")

# 展示结果
result_df.orderBy("time_is_seconds").show()

代码说明 #

• is_below_5：标记行是否满足value<5，为后续判断连续区间提供依据
• group_id：通过窗口函数累加，当遇到新的value<5连续区间时，group_id递增，确保同一连续区间的行归为同一组
• 分组聚合：对每个group_id计算对应的min和max时间，关联回原表后，非符合条件的行自动保留Null值

内容的提问来源于stack exchange，提问作者user8178045