NiFi 1.9.2：如何阻止QueryDatabaseTable重复执行致MongoDB重复插入

嘿，我对NiFi 1.9.2的QueryDatabaseTable组件挺熟悉的，你遇到的重复插入问题其实是这个组件的常见场景，不过有几个靠谱的方案能从源头或者兜底层面解决，咱们一个个说：

1. 用跟踪列（Tracking Column）实现增量同步（最推荐） #

这是QueryDatabaseTable设计的核心功能，专门用来避免重复拉取数据。你的问题根源就是没配置跟踪列，导致每次调度都全量查询100条记录：

• 打开QueryDatabaseTable的配置面板，找到Maximum Value Columns选项，选择一个具备递增/唯一特性的列，比如表的主键employee_id（自增主键最佳）或者记录的更新时间列last_updated
• 设置Initial Maximum Value：如果选的是employee_id，就填当前表中最大的id值（比如当前最大是100，就填100）；如果是时间列，填你开始同步的起始时间（比如2024-01-01 00:00:00）
• 关键：修改你的自定义查询，把跟踪列的过滤条件加进去。比如原来的select * from employee要改成：

  select * from employee where employee_id > ?
  

  NiFi会自动把?替换成上次执行后记录的最大employee_id值，这样每次只会拉取新增的记录，不会重复拉取旧数据

2. 利用状态管理自定义同步逻辑 #

如果你的表没有合适的跟踪列（比如没有自增主键或时间戳列），可以通过NiFi的状态管理来手动维护同步状态：

• 确保QueryDatabaseTable启用了状态管理（单机模式默认已启用，集群模式需要配置分布式状态存储）
• 在自定义查询中加入基于状态的过滤条件，比如假设你用一个自定义的同步标记：

  select * from employee where sync_flag = 0
  

• 执行完同步后，添加一个ExecuteSQL处理器，把已同步的记录的sync_flag设为1，同时在QueryDatabaseTable中配置状态变量，记录上次同步的范围
• 这个方法相对复杂，适合特殊场景，优先推荐第一种方案

3. 调整调度策略为单次执行（仅适用于一次性同步） #

如果你的需求只是一次性把数据从Oracle同步到MongoDB，不需要定期增量同步：

• 把QueryDatabaseTable的Run Schedule设为0，然后手动触发一次执行即可
• 或者用CRON表达式设置成只运行一次的时间点，比如0 0 12 1 1 ?（每年1月1日中午12点执行一次）

4. MongoDB端设置唯一索引兜底 #

如果前面的方案暂时无法实施，可以在MongoDB层面做去重兜底：

• 给MongoDB的employee集合添加唯一索引，比如基于employee_id：

  db.employee.createIndex({employee_id: 1}, {unique: true})
  

• 这样即使NiFi重复发送数据，MongoDB会自动拒绝重复的记录（不会抛出致命错误，具体可以在PutMongoRecord的错误处理策略中配置）
• 注意：这是兜底方案，最好还是从NiFi端阻止重复拉取，避免浪费资源

额外注意事项 #

NiFi 1.9.2的QueryDatabaseTable在使用自定义查询时，必须手动把跟踪列的过滤条件加到SQL里，不像新版本会自动拼接，所以一定要记得修改你的自定义SQL语句哦！

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