如何用Apache NiFi实现Oracle到SQL Server关联表高效ETL？

核心实现流程 #

先完成Header表的增量同步，再将每条Header记录的ID提取为FlowFile属性，以此作为参数查询对应Detail数据并同步，具体步骤如下：

1. 提取Header记录的ID到FlowFile属性

   • 在SplitAvro拆分单条Header记录后，不能仅用FilterAttribute（它只能保留已有属性，无法从内容中提取新属性），改用以下方式：
     • 若FlowFile为Avro格式：用ConvertRecord将Avro转为JSON，再添加EvaluateJsonPath处理器，设置Destination为flowfile-attribute，属性名设为header_id，值填$.ID（需替换为实际Avro字段名），即可把内容中的ID提取为FlowFile属性。
     • 若为其他格式：用ExtractText处理器，根据数据格式编写正则表达式捕获ID值，输出到指定属性。
   • 用AttributeExists处理器过滤未成功提取ID属性的FlowFile，避免后续报错。

2. 配置ExecuteSQL关联查询Detail数据

   • 在ExecuteSQL的SQL Select Query中直接引用属性作为参数：

     select * from Detail d where d.header_id = ${header_id}
     

   • 确保处理器的Database Connection Pooling Service正确指向Oracle源库，且属性名header_id与SQL变量完全一致。

3. 批量写入Detail数据到SQL Server

   • ExecuteSQL输出的Detail结果集直接接入PutDatabaseRecord处理器，配置好SQL Server连接池及对应Record Reader/Writer，批量写入Detail表即可。

性能优化建议 #

• 调整QueryDatabaseTable的Max Rows Per Flow File和SplitAvro的Batch Size，平衡单FlowFile数据量与处理速度。
• 给ExecuteSQL和PutDatabaseRecord设置合理并发线程数，避免数据库压力过载。
• 保留Header增量同步逻辑，仅处理新增/更新的Header，减少不必要的Detail查询。

排查要点 #

• 用LogAttribute处理器打印FlowFile属性列表，确认ID属性是否正确提取。
• 查看ExecuteSQL日志，检查SQL语句是否正确替换参数值。

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