Structured Streaming读GKE Strimzi Kafka写Mongo去重方案咨询

前置说明 #

你贴的现有代码本身无法正常运行：readStream返回的是流式DataFrame，不支持直接调用批处理的write()方法，提交就会报错，以下回答会同步覆盖该问题的修正方案。

问题1：定时批任务如何避免重复读Kafka、重复写入Mongo #

核心要做两层保障，缺一不可：

• 第一层：持久化管理消费偏移量
  不要依赖临时集群本地存储、也不要依赖Kafka消费组自动提交偏移量——Spark读Kafka时默认不会向Kafka broker提交消费位移，你配置的kafka.group.id仅作监控标识用，根本不会帮你记录消费位置。
  你需要选一个持久化存储（GCS最适配你的GCP技术栈，也可以直接在Mongo里建一个专门存偏移量的集合），每次任务启动时先读取上一次任务成功后记录的各Topic分区偏移量，作为本次消费的起始位置；等本次所有数据成功写入Mongo后，再把本次消费到的最新分区偏移量写回持久化存储，供下次任务使用。
• 第二层：写入端做幂等保障
  给每条数据生成全局唯一主键：优先用消息自带的业务唯一ID，没有的话直接用Topic名+分区号+偏移量拼接即可，在Mongo侧对这个主键建唯一索引，写入时用upsert模式，遇到重复主键直接跳过/覆盖。就算任务异常重跑、偏移量记录有偏差，也不会产生重复数据。

问题2：readStream默认读取行为、spark.read与spark.readStream的核心差异 #

• readStream读Kafka的默认行为：不会自动从上次消费位置续读。
  如果你不配置checkpointLocation，或者配置的路径下没有历史偏移量记录，就会按你设置的startingOffsets参数启动：默认值是latest（仅读任务启动后新进入Topic的数据），你现在代码里写的是earliest，每次启动都会从Topic最开始的全量数据读。只有当checkpoint路径下存在有效历史偏移量时，readStream才会从上次停止的位置续读——但你每次跑完就删除Dataproc集群，如果checkpoint存在本地或临时集群存储上，肯定会丢失，自然每次都会从头读。
• 该场景下spark.read（批式读Kafka）和spark.readStream（流式读Kafka）的核心差异：
  • spark.read是纯批语义，必须手动指定要读取的各分区起始、结束偏移量范围，提交后Spark会一次性把指定范围的数据读完，作业直接结束，不会持续等待新数据，天然适配定时跑批、跑完就销毁集群的场景。
  • spark.readStream默认是常驻流式语义，会持续监听Kafka新数据并运行，除非配置Trigger.Once()（老版本）或Trigger.AvailableNow()（Spark 3.3+），才会在消费完当前可获取的所有数据后自动停止作业。如果不配这两个触发器，readStream启动的作业会一直挂起，完全不适合你的调度逻辑。

问题3：Mongo不支持Structured Streaming流式写入、无法直接用checkpoint的最优方案 #

结合你的GCP技术栈、临时集群、10分钟定时调度的场景，最优方案是纯批式读取+手动管理偏移量+Mongo幂等写入，完全没必要硬套Structured Streaming的流模式，落地步骤如下：

1. 废弃现有readStream写法，改用spark.read.format("kafka")批式API：
   • 任务启动后先从持久化存储（GCS/ Mongo偏移量集合）读取上一次记录的各分区偏移量，作为本次读取的startingOffsets
   • 先获取目标Topic当前所有分区的最新偏移量，作为本次读取的endingOffsets
   • 把起始、结束偏移量传入read配置，只读这一段固定范围的数据，不会多取也不会漏取
2. 数据处理逻辑保留你现有的实现：解析JSON、过滤对应customer的数据即可，额外给每条数据拼接全局唯一主键。
3. 写入Mongo时用upsert模式，基于唯一主键做幂等写入，避免重复数据。
4. 等所有数据写入Mongo的确认返回后，再把本次用到的endingOffsets更新到持久化存储中，之后作业即可结束，直接删除Dataproc集群。

这个方案的优势很明显：

• 完全不依赖Spark checkpoint，集群销毁不会影响状态存储，也不会出现checkpoint小文件膨胀的问题
• 偏移量逻辑完全透明，出问题时可以手动修改偏移量重跑指定范围的数据，运维成本极低
• 批作业跑完自动退出，不会出现流式作业意外挂住浪费资源的问题
• 偏移量管理+幂等写入的组合可以实现至少一次消费语义，就算任务异常中断，也不会丢数据或产生重复数据。

如果你一定要用Structured Streaming，次优方案是把checkpoint存在持久化GCS路径上，配置Trigger.AvailableNow()触发器，通过foreachBatch自定义逻辑，在每个批次内调用Mongo的批式写入API写数据，等批次写入成功后再提交偏移量。但这个方案整体更重，checkpoint维护成本高，远不如纯批方案适配你的场景。

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