如何避免Spark Structured Streaming留存旧数据引发内存溢出问题

• 无水印配置的全局groupBy确实会永久留存所有分组的状态数据，随着运行时间增长必然引发OOM，和你的预判一致。
• 基于消息接收时间加虚拟水印是可行方案，但并非唯一解决方案：
  • 方案1（跨批次聚合场景）：你可以直接复用Kafka消息自带的timestamp字段（Kafka服务端写入时间，也可以自定义消费时的处理时间列）作为事件时间列，设置水印阈值大于单条消息拆包、聚合的最大耗时即可，比如确认单条消息处理不会超过10分钟，就配置withWatermark("msg_time", "10 minutes")，同时将msg_time按合适的粒度（比如1分钟窗口）加入groupBy条件，到期后Spark会自动清理对应message-id的状态，不会出现状态无限膨胀的问题。
  • 方案2（单批次内即可完成聚合场景）：如果可以保证同一个message-id只会出现在同一个微批次中，完全不需要用全局流式聚合，改用mapGroups/flatMapGroups在单批次内做本地聚合即可，这种逻辑不会产生全局状态，完全不存在内存留存问题，性能也更优。
• 注意Spark 2.4.7版本要求水印必须对应groupBy中的时间列，仅加水印不加入分组条件的话，水印规则不会生效。

• 这种场景完全不需要加任何水印、也不需要做额外的groupBy操作：只要你的处理逻辑中没有用到聚合、流表join、自定义状态算子这类需要持久化状态的算子，Structured Streaming根本不会留存任何历史数据，每个微批次处理完成后就会直接释放对应内存，不会出现内存无限增长的问题。
• 你可以直接用map/flatMap/explode等转换算子处理每条消息，处理完成后直接写入下游即可，和DStream的单条处理逻辑开销没有本质区别，还能保留Structured Streaming的精确一次语义、SQL API兼容等优势。
• 只有当你用到需要维护状态的算子（比如agg、groupBy、流-流join、mapGroupsWithState、flatMapGroupsWithState等）时，才需要考虑状态清理和水印配置，纯无状态流处理不需要做任何额外适配。

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