基于事件驱动自动化NiFi流：MongoDB到PostgreSQL批量迁移问题

针对你手动分批拉取1亿条Mongo文档到PostgreSQL的痛点，直接改造现有流程即可实现全自动化，核心思路是断点持久化+循环触发，具体步骤如下：

1. 替换GenerateFlowFile：自动获取同步断点 #

删掉GenerateFlowFile，改用QueryDatabaseTable组件从PostgreSQL获取上次同步的最后一条MongoDB _id：

• 配置PostgreSQL连接池，编写SQL查询断点：

  SELECT last_mongo_id FROM nifi_sync_breakpoint LIMIT 1
  

  第一次同步前先创建断点表：CREATE TABLE nifi_sync_breakpoint (last_mongo_id VARCHAR(255) PRIMARY KEY)，初始可插入空值或让查询返回null。
• 组件输出的FlowFile会携带last_mongo_id属性，作为后续Mongo查询的起点。

2. 改造GetMongo：实现分页拉取 #

配置GetMongo组件，基于断点拉取指定数量的文档：

• Query字段写动态条件：

  ${last_mongo_id:isEmpty():ifElse({}, {"_id": {"$gt": "${last_mongo_id}"}})}
  

  第一次同步时last_mongo_id为空，拉取全量数据的前500k；后续只拉取比上次断点新的文档。
• 设置Limit为500000，强制每次拉取不超过500k条。
• 开启Query Batch Mode，确保MongoDB按批次返回数据，避免内存溢出。

3. 持久化本次同步的断点 #

在ExecuteSQL插入PostgreSQL之后，新增两个组件完成断点更新：

• 先用QueryRecord组件从当前FlowFile中提取最大_id：
  • 编写SQL：SELECT MAX(_id) AS current_max_id FROM FLOWFILE
  • 输出的FlowFile会携带current_max_id属性。
• 再用ExecuteSQL执行断点更新：

  INSERT INTO nifi_sync_breakpoint (last_mongo_id) VALUES (?) ON CONFLICT (last_mongo_id) DO UPDATE SET last_mongo_id = EXCLUDED.last_mongo_id
  

  将current_max_id作为参数传入，实现原子更新，避免并发冲突。

4. 实现自动循环触发 #

在断点更新完成后，新增RouteOnAttribute组件判断是否需要继续同步：

• 添加路由规则：
  • 规则名：need_continue
  • 规则表达式：${record_count:equals(500000)}（record_count来自QueryRecord统计的本次同步记录数）
• 如果规则成立（本次拉满500k条，说明还有数据），就把FlowFile路由回QueryDatabaseTable组件触发下一轮同步；如果不成立（拉取数量不足500k，说明已同步完所有数据），则结束流程。

最终流程结构 #

QueryDatabaseTable → GetMongo → ConvertJSONtoSQL → ExecuteSQL → QueryRecord(统计记录数+取最大_id) → ExecuteSQL(更新断点) → RouteOnAttribute → [继续则回到QueryDatabaseTable，否则结束]

关键注意事项 #

• 确保MongoDB的_id是有序的（比如默认的ObjectId，基于时间戳生成，天然有序），否则$gt条件无法正确拉取增量数据。
• 给NiFi组件配置足够内存，避免处理500k条文档时出现OOM。
• 新增LogAttribute组件记录每次同步的last_mongo_id和record_count，方便排查问题。

内容的提问来源于stack exchange，提问作者hoa.le