如何基于SQL表数据在NiFi中动态过滤JSON格式FlowFile

需求说明 #

基于CustomerID字段过滤JSON格式的FlowFile，仅保留指定SQL表中存在对应CustomerID的记录。待过滤FlowFile示例：

[ { "CustomerID" : 1, "OrderNbr" : "12", "Amount":100 } ]

解决方案流程 #

原QueryRecord的静态WHERE条件无法直接关联SQL表数据，需调整流程如下：

1. 拉取SQL表中的有效CustomerID

   • 使用QueryDatabaseTable处理器，配置目标数据库连接，SQL语句设为：

     SELECT DISTINCT CustomerID FROM 你的目标表名
     

   • 用ConvertRecord处理器将查询结果转换为JSON数组格式（如[{"CustomerID":1},{"CustomerID":3},...]），确保后续处理器可解析。

2. 关联过滤待处理FlowFile
   推荐使用LookupRecord处理器实现动态匹配：

   • RecordReader：选择JsonTreeReader，配置解析待过滤JSON的规则（如设置根节点为数组）。
   • RecordWriter：选择JsonRecordSetWriter，输出过滤后的JSON结果。
   • Lookup Service：配置DatabaseLookupService，连接目标数据库，Lookup Query设为：

     SELECT CustomerID FROM 你的目标表名 WHERE CustomerID = ?
     

   • 核心配置：设置Lookup Column为待过滤FlowFile中的CustomerID字段，Output Column命名为MatchedID。
   • 过滤逻辑：添加QueryRecord处理器，执行查询语句保留匹配成功的记录：

     SELECT * FROM FLOWFILE WHERE MatchedID IS NOT NULL
     

   替代方案：使用JoinRecord处理器，将待过滤FlowFile与QueryDatabaseTable的输出做INNER JOIN（关联字段为CustomerID），自动保留两边匹配的记录，后续用QueryRecord剔除多余字段即可。

原方案问题说明 #

原QueryRecord的SELECT * FROM FLOWFILE WHERE ${key} = ${value}语句依赖静态属性变量，无法动态匹配SQL表中的所有有效CustomerID，必须先拉取有效ID集合再做关联过滤。

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