如何验证CDC数据管道？数据防丢失与丢失检测行业标准方案

现有CDC管道链路：MongoDB数据源 → 自定义Python CDC消费程序 → CDC数据落盘为文件 → Spark读取文件执行SQL计算 → 计算结果写入Kafka

一、全链路无数据丢失的保障方案 #

整条链路要实现零数据丢失，核心是每个节点都遵守「下游持久化成功后，再提交上游消费位点」的原则，每个环节的具体要求如下：

• MongoDB源端配置
  开启Change Stream的fullDocument输出，将oplog保留窗口从默认的几小时调整到至少72小时，避免下游消费积压过久导致resume token失效。消费位点（resume token）必须持久化到独立的可靠存储（比如专门的MongoDB集合、多副本磁盘），禁止消费程序重启时直接从最新时间点开始消费。
• Python消费落盘环节
  这是自定义CDC链路最容易出丢数问题的环节，必须做两阶段提交：
  1. 消费到CDC数据后先写入.tmp后缀的临时文件，写完执行fsync强制将数据刷到物理磁盘，不能只写到操作系统页缓存就认为写入成功；
  2. 将临时文件原子重命名为正式的.cdc后缀文件，只有重命名成功后，才更新持久化存储里对应批次的resume token。
     绝对不能图省事先提交resume token再写文件，一旦进程在提交token后、文件刷盘前崩溃，这部分数据会彻底丢失，且无法从源端补拉。另外存储CDC文件的介质必须用多副本可靠存储（RAID盘、HDFS、多副本对象存储都可以），禁止用单节点无备份本地盘存核心CDC数据。
• Spark计算环节
  不管是离线批处理还是微批流处理，都必须开启可靠的checkpoint机制，checkpoint目录存到多副本可靠存储，记录每个已处理文件的偏移量，禁止手动删除checkpoint目录。源文件不能在读取后立刻删除，必须等当前批次计算完成、下游写入成功后，再做归档或删除操作，任务失败重跑时要能完整重跑未提交成功的批次。另外要合理配置executor内存、开启任务失败重试，shuffle阶段开启中间结果多副本存储，避免OOM、节点故障导致的数据丢失。
• Kafka写入环节
  生产者必须配置acks=all、enable.idempotence=true，重试次数设为足够大的值（配合指数退避重试策略），禁止为了吞吐量用acks=0或acks=1配置。只有收到Kafka集群返回的写入成功ack后，Spark任务才能标记当前批次处理完成，提交对应checkpoint。Kafka集群侧要配置至少3副本，min.insync.replicas=2，避免单节点故障导致已写入数据丢失。

二、数据丢失的检测与精准定位方案 #

靠日志排查丢数问题效率极低，必须提前在链路里埋好可观测指标：

• 前置埋点规则
  直接用MongoDB oplog自带的全局单调递增ts时间戳作为每个CDC事件的唯一有序序列号，在每个链路节点（落盘完成、Spark读取完成、计算完成、Kafka写入完成）按分钟粒度上报三个核心指标：当前节点处理到的最大序列号、当前分钟处理的事件总条数、对应批次的处理状态。
• 实时检测逻辑
  正常链路下，下游节点的最大序列号和上游的差值不会超过配置的延迟阈值（比如10分钟），相邻节点同时间窗口的事件条数差应该趋近于0。一旦出现下游序列号长时间停滞、相邻节点事件数差值超过阈值、端到端（源端1小时产生的CDC总条数 vs Kafka写入1小时总条数）对账差值不为0，立刻触发丢数告警。
• 点位定位方法
  告警触发后从上游到下游逐节点对比序列号范围和事件计数即可快速定位：
  1. 先对比MongoDB源端指定时间范围的oplog序列号范围，和对应时间落盘文件里的序列号范围，如果存在序列号缺口，说明丢在Python消费环节，直接查对应时间点的resume token提交记录、程序崩溃日志、磁盘IO日志就能定位根因；
  2. 如果落盘文件的序列号连续、事件数和源端一致，再对比落盘文件总条数和Spark任务的输入记录数，如果有缺口，说明丢在文件读取环节，查Spark文件扫描列表、checkpoint记录，排查是否存在漏读文件、文件损坏问题；
  3. 如果Spark输入记录数和落盘文件一致，再对比Spark计算输出条数和Kafka返回的写入成功条数，如果有缺口，说明丢在计算或Kafka写入环节，查Spark任务失败日志、Kafka生产者报错日志即可定位。
• 每个环节都要保留批次级审计日志，记录每个批次处理的文件路径、序列号起止范围、事件数、处理状态，日志至少保留30天，排查时不需要全量扫数据，直接按序列号查对应批次日志即可。

三、通用处理范式与行业落地方案 #

• 通用处理范式没有特殊黑科技，核心就是两点：一是全链路实现至少一次语义，所有节点严格遵守两阶段提交原则，从机制上避免丢数；二是全链路埋入有序可对账的水位指标，做实时端到端对账，即使出现磁盘损坏、集群级故障等极端情况，也能第一时间发现缺口、定位到具体故障节点，再通过重放备份的CDC数据补全缺口。
• 行业常见的成熟落地方案：
  • 源端CDC消费尽量不要从零写自定义Python脚本，成熟方案是用已经经过大规模生产验证的CDC组件对接MongoDB Change Stream，自带位点持久化、重试、容错逻辑，可靠性远高于自定义代码；
  • 中间转储环节如果想降低文件存储带来的可靠性复杂度，可以直接将CDC数据先写入Kafka做缓冲，再用Spark消费Kafka数据做计算，省去自定义文件落盘的两阶段提交逻辑，Kafka本身的多副本机制可靠性远高于自定义文件存储；
  • 计算环节优先用Structured Streaming，原生支持可靠checkpoint和端到端语义，不需要自己实现位点管理、批次提交逻辑，减少自研代码带来的可靠性风险；
  • 数据对账环节可以用成熟的数据质量工具做行数校验、主键连续性校验，替代自研监控脚本，降低漏告警概率。

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