Apache Spark结构化流：能否记录水印丢弃事件及是否有监听接口？

好问题！在Apache Spark结构化流场景下，我们确实可以追踪因水印机制被丢弃的晚到事件，但默认Spark不会主动帮你记录这些信息，得靠特定配置和自定义逻辑来实现；另外，目前并没有专门暴露的监听器接口直接获取这些被丢弃的事件，不过有间接的方式可以监控相关状态变化。

一、如何记录被水印丢弃的具体事件 #

这里分两种场景来处理：

1. 利用Spark内置配置监控状态清理（Spark 3.1+） #

从Spark 3.1版本开始，你可以开启配置项 spark.sql.streaming.stateStore.log.deletedState，将其设置为true。开启后，Spark会把被水印清理的状态条目相关信息（比如状态的键、清理时间等）记录到状态存储的日志中。不过要注意，这是状态层面的日志，并不是直接记录原始的晚到事件内容，适合用来监控状态清理的整体情况。

2. 自定义分流逻辑记录原始晚到事件 #

如果需要追踪具体的晚到事件内容，最可靠的方式是在数据流中添加分支处理：

• 首先对数据流应用withWatermark设置水印；
• 然后使用current_watermark()内置函数，判断事件的事件时间是否晚于当前水印值，筛选出这些晚到事件；
• 最后将这部分分流出来的事件写入专门的输出Sink（比如文件系统、Kafka、数据库等），实现持久化记录。

举个简单的代码示例：

import org.apache.spark.sql.functions.current_watermark

// 原始数据流
val rawStream = spark.readStream.format("kafka").load()
  .select(from_json(col("value").cast("string"), schema).alias("data"))
  .select("data.eventTime", "data.*")
  .withWatermark("eventTime", "10 minutes")

// 正常处理的数据流
val processedStream = rawStream.filter(col("eventTime") >= current_watermark("eventTime"))
  .writeStream.format("parquet").option("path", "/path/to/processed").start()

// 被水印丢弃的事件分流
val droppedEventsStream = rawStream.filter(col("eventTime") < current_watermark("eventTime"))
  .writeStream.format("json").option("path", "/path/to/dropped-events").start()

这种方式能完整保留每一个被水印过滤的原始事件，但会带来一定的计算开销，需要根据你的业务场景权衡资源消耗。

二、关于监听器接口的说明 #

目前Spark结构化流并没有提供专门的监听器接口来直接监听“水印丢弃事件”，不过可以通过StreamingQueryListener间接获取相关的状态清理统计信息：

• 实现StreamingQueryListener接口，重写onQueryProgress方法，在该方法中可以获取QueryProgress对象；
• 从QueryProgress里能拿到当前的水印值（progress.watermark），以及状态清理的统计数据（比如progress.stateOperators中的numRowsRemovedByWatermark字段，代表被水印清理的状态条目数）。

但要注意，这种方式只能拿到统计数据，无法获取具体的被丢弃事件内容，如果需要具体事件，还是得用前面提到的自定义分流方法。

注意事项 #

• 对于Spark 3.1之前的版本，spark.sql.streaming.stateStore.log.deletedState配置不存在，只能依赖自定义分流逻辑来记录晚到事件；
• 自定义分流时，要确保水印已经生效，否则current_watermark()返回的是null，会导致过滤逻辑出错。

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