Apache Spark窗口运算输出无序问题的解决方案咨询

我用Python开发了一个Spark Streaming应用，从Kafka读取事件数据，通过窗口操作（Window Operation）和水印（Watermark）做聚合后输出到控制台，但输出结果的顺序不符合预期。

事件数据示例 #

{"id": 292, "stock": "g", "price": 817, "time": "2023-03-06 17:32:52.596087"}

聚合代码 #

# 将JSON数据转换为结构化数据
df_select = df.selectExpr("CAST(value AS STRING)"). \
  select(from_json("value", schema).alias("data")). \
    select("data.*")

# 基于time字段做窗口切割，按stock分组聚合
df_agg = df_select.withWatermark("time", "10 seconds") \
.groupBy(
  window("time", "10 seconds"),
  "stock"
).agg(F.min("price").alias("min_price"), F.max("price").alias("max_price")) 

输出乱序问题 #

输出结果存在无序记录（示例如下），已知流数据不能直接做全局排序，请问该如何解决？

+------------------------------------------+-----+---------+---------+
|window                                    |stock|min_price|max_price|
+------------------------------------------+-----+---------+---------+
|{2023-03-06 17:28:50, 2023-03-06 17:29:00}|e    |475      |475      |
|{2023-03-06 17:28:50, 2023-03-06 17:29:00}|i    |917      |917      |
|{2023-03-06 17:28:50, 2023-03-06 17:29:00}|m    |538      |538      |
|{2023-03-06 17:28:50, 2023-03-06 17:29:00}|r    |270      |270      |
|{2023-03-06 17:28:50, 2023-03-06 17:29:00}|f    |253      |253      |
|{2023-03-06 17:28:50, 2023-03-06 17:29:00}|j    |280      |280      |
|{2023-03-06 17:28:40, 2023-03-06 17:28:50}|i    |503      |775      |
|{2023-03-06 17:28:50, 2023-03-06 17:29:00}|c    |834      |834      |
+------------------------------------------+-----+---------+---------+

解决办法 #

流数据乱序的核心原因是迟到数据触发旧窗口的聚合更新，导致旧窗口结果穿插在新窗口结果之后输出。可以从以下几个方向处理：

1. 微批内按窗口时间排序 #

虽然流数据不支持全局排序，但可以在每个输出的微批内，基于窗口起始时间做局部排序，让当前微批的结果是有序的。修改输出逻辑：

# 在输出前对微批内数据按窗口起始时间、stock排序
df_agg.orderBy("window.start", "stock").writeStream \
  .outputMode("update") \
  .format("console") \
  .start() \
  .awaitTermination()

注意：如果后续还有迟到数据触发旧窗口更新，这条更新记录仍会出现在后续微批中。

2. 调大水印延迟匹配业务场景 #

当前窗口大小10秒，水印也是10秒，意味着窗口结束后10秒就会关闭。如果业务允许，可以适当调大水印延迟，让更多迟到数据在窗口关闭前到达，减少后续的窗口更新输出：

df_agg = df_select.withWatermark("time", "15 seconds") \
.groupBy(
  window("time", "10 seconds"),
  "stock"
).agg(F.min("price").alias("min_price"), F.max("price").alias("max_price")) 

3. 改用Append输出模式（适合最终一致性场景） #

如果不需要实时看到窗口的中间聚合结果，只需要窗口的最终结果，可以切换到append输出模式。这种模式下，只有当窗口被水印彻底关闭（不会再收到新数据）时，才会输出该窗口的最终结果，输出天然按窗口时间顺序：

df_agg.writeStream \
  .outputMode("append") \
  .format("console") \
  .start() \
  .awaitTermination()

注意：Append模式要求聚合字段都是确定性的，且依赖水印机制确保窗口不会再收到数据。

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