如何用NiFi内置处理器从嵌套JSON提取仅变更属性生成新JSON

我之前也碰到过类似的JSON结构转换需求，NiFi里用JoltTransformJSON处理器就能完美搞定，它专门用来处理JSON的结构变换，完全匹配你要保留顶层属性、只提取columns中值有变化字段的需求。

具体实现步骤： #

1. 添加并配置JoltTransformJSON处理器
   在NiFi画布中拖入JoltTransformJSON处理器，核心是配置Jolt Specification（转换规则），下面是针对你需求的完整规则：

   [
     {
       "operation": "modify-overwrite-beta",
       "spec": {
         "filteredColumns": "=filter(@(1,columns), function($) { $.last_value != $.value })"
       }
     },
     {
       "operation": "shift",
       "spec": {
         "*": "&", // 原样保留所有顶层属性（type、database等）
         "filteredColumns": {
           "*": {
             "name": {
               "*": {
                 "@(2,value)": "&" // 将符合条件的列的name作为键、value作为值映射到顶层
               }
             }
           }
         }
       }
     },
     {
       "operation": "remove",
       "spec": {
         "columns": "", // 移除原始的columns数组
         "filteredColumns": "" // 移除临时生成的filteredColumns字段
       }
     }
   ]
   

2. 规则逻辑拆解

   • 第一步modify-overwrite-beta：用NiFi内置的filter函数筛选出columns数组里last_value和value不相等的项，存到临时字段filteredColumns中。
   • 第二步shift：先把所有顶层原有属性（比如type、database这些）保留下来，再把filteredColumns里的每个项，以name为键、value为值，映射到顶层JSON结构里。
   • 第三步remove：删掉不需要的原始columns数组和临时的filteredColumns字段，得到干净的最终结构。

3. 测试验证
   你可以用GenerateFlowFile处理器生成你提供的示例输入JSON，连接到JoltTransformJSON，再用LogAttribute处理器查看输出，就能得到预期结果：

   { "type" : "update", "database" : "ABC", "table_name" : "EXT", "table_id" : 78, "TRS": "FRD" }
   

这个方案完全用NiFi内置处理器实现，不需要额外脚本，后续维护也很方便。

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