PySpark 连续时间窗口分组聚合实现：空窗口补0及多维度适配

原生PySpark的窗口聚合逻辑仅会输出存在匹配数据的窗口，补全无数据窗口的核心思路为：先生成全量连续时间窗口的基准数据集，再与实际聚合结果做左连接，将空值替换为0，最终输出连续的时间序列结果。

基础无维度分组场景实现 #

针对无额外分组维度、仅按时间窗口聚合的需求，实现步骤如下：

• 从原始数据中提取全局最早/最晚时间，向下/向上取整到整小时，作为连续窗口的生成边界
• 按指定的60分钟步长，基于起止边界生成所有连续时间窗口，构造基准数据集
• 计算实际的窗口聚合结果，与基准数据集按窗口做左连接，将空聚合值替换为0

对应实现代码：

import pyspark.sql.functions as sf
from pyspark.sql.types import TimestampType

# 转换原始时间字段为Timestamp类型
df1 = df.select(
    sf.unix_timestamp(sf.col("datetime"), 'yyyy/MM/dd HH:mm:ss').cast(TimestampType()).alias("timestamp"),
    sf.col("value")
)

# 计算全局窗口的起止边界
time_bound = df1.agg(
    sf.floor(sf.min("timestamp").cast("long") / 3600).cast("timestamp").alias("start_bound"),
    sf.ceil(sf.max("timestamp").cast("long") / 3600).cast("timestamp").alias("end_bound")
).collect()[0]

# 生成连续的60分钟时间窗口序列
window_base = sf.sequence(
    time_bound["start_bound"],
    time_bound["end_bound"],
    sf.expr("INTERVAL 1 HOUR")
).select(
    sf.struct(
        sf.col("value").alias("start"),
        (sf.col("value") + sf.expr("INTERVAL 1 HOUR")).alias("end")
    ).alias("window")
)

# 计算实际的窗口聚合值
actual_agg = df1.groupBy(sf.window(sf.col("timestamp"), "60 minutes")).agg(
    sf.sum("value").alias("sum(value)")
)

# 左连接补全空窗口，空值替换为0
final_result = window_base.join(
    actual_agg,
    on="window",
    how="left"
).fillna(0.0, subset=["sum(value)"]).orderBy("window.start")

final_result.show(truncate=False)

运行后输出将和期望的基础场景效果完全一致，自动补上中间无数据的04:00-05:00窗口，聚合值为0.0。

带name维度的多分组场景实现 #

针对需要按name分组、每个name下都覆盖全时间范围连续窗口的需求，实现逻辑调整为：

• 提取数据中所有不重复的name值，作为分组维度的全集
• 复用上述全量连续窗口的生成逻辑
• 将name全集与全量窗口做笛卡尔积，构造「name + 时间窗口」的基准数据集，保证每个name都覆盖所有时间窗口
• 计算带name维度的实际窗口聚合结果，与基准数据集按「name + 窗口」做左连接，空聚合值替换为0

对应实现代码：

# 转换原始数据字段
df1 = df.select(
    "name",
    sf.unix_timestamp(sf.col("datetime"), 'yyyy/MM/dd HH:mm:ss').cast(TimestampType()).alias("timestamp"),
    sf.col("value")
)

# 提取所有不重复的name全集
name_set = df1.select("name").distinct()

# 计算全局窗口起止边界
time_bound = df1.agg(
    sf.floor(sf.min("timestamp").cast("long") / 3600).cast("timestamp").alias("start_bound"),
    sf.ceil(sf.max("timestamp").cast("long") / 3600).cast("timestamp").alias("end_bound")
).collect()[0]

# 生成连续的60分钟时间窗口序列
window_base = sf.sequence(
    time_bound["start_bound"],
    time_bound["end_bound"],
    sf.expr("INTERVAL 1 HOUR")
).select(
    sf.struct(
        sf.col("value").alias("start"),
        (sf.col("value") + sf.expr("INTERVAL 1 HOUR")).alias("end")
    ).alias("window")
)

# 构造name与全窗口的笛卡尔积基准数据集
base = name_set.crossJoin(window_base)

# 计算带name维度的实际窗口聚合值
actual_agg = df1.groupBy(
    "name",
    sf.window(sf.col("timestamp"), "60 minutes")
).agg(sf.sum("value").alias("sum(value)"))

# 左连接补全空记录，空值替换为0
final_result = base.join(
    actual_agg,
    on=["name", "window"],
    how="left"
).fillna(0.0, subset=["sum(value)"]).orderBy("name", "window.start")

final_result.show(truncate=False)

运行后将输出每个name下从全局最早窗口到最晚窗口的全量连续记录，无数据窗口的聚合值自动补为0.0，完全匹配扩展场景的预期效果。

注意事项

• 生成的窗口边界与原生sf.window的整点对齐规则完全一致，不会出现窗口错位导致的聚合值不匹配问题
• 若业务需要固定的时间范围（而非从数据中自动提取起止时间），可直接替换time_bound中的起止值，无需依赖原始数据的时间分布

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