PySpark Streaming窗口聚合仅展示窗口起始时间的实现方法

你无需将聚合结果存入临时DataFrame做后续拆分，直接利用窗口函数返回结构体的内置start字段调整分组维度即可，全程支持实时流式输出，改写代码如下：

import pyspark.sql.functions as F

stream = streaming.selectExpr("car", "cost", "timestamp")\
        .withWatermark("timestamp", "30 seconds")\
        .groupBy(
            F.col("car"),
            # 直接提取窗口起始时间作为分组维度，无中间结果留存
            F.window("timestamp", "1 hour")["start"].alias("tmst_window")
        )\
        .agg(F.sum("cost").alias("agg_cost"))

实现说明 #

• 1小时窗口函数的返回值为结构化类型，原生包含start（窗口起始时间）、end（窗口结束时间）两个内置字段
• 直接在分组逻辑中提取start字段并重命名，和原有计算逻辑完全兼容，无额外shuffle开销
• 整个计算流程嵌入流式增量计算链路，不会生成中间结果DataFrame，输出的tmst_window直接为时间戳类型，完全匹配目标输出格式
• 原有水印逻辑未做改动，状态清理、超时触发规则和原实现完全一致，不会引入计算偏差

最终输出效果 #

+--------------+---------------------+------------------+
|car           |tmst_window          |agg_cost          |
+--------------+---------------------+------------------+
|Toyota        |2010-12-01 14:00:00 |10                |
|Ford          |2010-12-01 14:00:00 |20                |
|Audi          |2010-12-01 13:00:00 |30                |
+--------------+---------------------+------------------+

如需自定义时间展示格式，可直接在字段上追加转换逻辑，例如F.date_format(F.window("timestamp", "1 hour")["start"], "yyyy-MM-dd HH:mm:ss").alias("tmst_window")，同样无需额外的中间处理环节。

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