NiFi流处理故障重启与数据不丢失方案咨询

核心原则 #

上线后禁止清除CaptureChangeMySQL的处理器状态，状态中记录了最后处理的binlog位点，是断点续传的关键。

方案1：优化CaptureChangeMySQL自身参数与状态管理 #

• 停止任务时绝不清除处理器状态，故障修复后直接启动处理器，它会自动从上次记录的binlog位置开始拉取故障期间的所有变更，无需从头全量拉取。
• 调整Max Batch Size参数，限制每次从binlog拉取的记录数（建议根据下游处理能力设为1000-5000），避免一次性涌入大量数据填满队列。
• 开启Back Pressure机制：设置队列的Back Pressure Object Threshold和Back Pressure Data Size Threshold，当队列达到阈值时，CaptureChangeMySQL会暂停拉取，直到队列数据被下游处理到阈值以下，防止队列溢出和OOM。

方案2：引入中间消息队列做持久化缓存 #

将CaptureChangeMySQL捕获的变更先写入持久化消息队列（如Kafka），再从队列消费到DWH：

• 使用PublishKafkaRecord处理器将变更数据发送到Kafka Topic，Kafka会持久化存储数据，支持按偏移量断点续传。
• 下游用ConsumeKafkaRecord处理器从Kafka消费并处理到DWH。即使NiFi故障，Kafka会保留故障期间的所有变更，修复后从上次的偏移量继续消费，完全避免数据丢失。
• 可通过Kafka的分区和消费者组机制实现并行处理，缓解队列积压，同时保证最新数据的处理效率。

方案3：保障MySQL Binlog的可访问性 #

• 配置MySQL的expire_logs_days参数，设置足够长的保留时间（建议至少7天，根据故障修复时长调整），确保故障期间产生的binlog不会被自动清理。
• 定期归档binlog到可靠存储（如本地磁盘阵列、对象存储），即使MySQL节点故障或binlog被误删，也能从归档中恢复所需的binlog文件，供NiFi补拉数据。

方案4：优化下游处理与队列优先级 #

• 使用PrioritizeAttribute处理器为变更记录添加优先级属性（比如基于timestamp字段，最新记录优先级更高），NiFi队列会优先处理高优先级数据，避免最新记录被积压在队列尾部。
• 下游用SplitRecord将大批次数据拆分为小批量，或用DistributeLoad处理器将数据分发到多个并行处理分支，提升整体处理吞吐量，减少队列积压。

方案5：NiFi集群高可用部署 #

将NiFi部署为集群模式，配置节点故障自动切换：

• 集群中的多个节点分担任务，单个节点故障时，其他节点会接管CaptureChangeMySQL任务，大幅缩短故障时间，减少数据丢失风险。
• 配合NiFi的Primary Node机制，确保CaptureChangeMySQL仅在主节点运行，避免重复拉取binlog。

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