如何在Nifi中使用JOLT动态替换数组内指定属性的同名同值字段

以下两种方案均支持动态扩展替换规则，所有匹配的同名属性都会被批量替换，满足需求：

方案1：使用JoltTransformJSON处理器（匹配你最初的技术选型） #

直接使用如下Jolt规范即可：

[
  {
    "operation": "shift",
    "spec": {
      "counterTop": {
        "loaf1": {
          "*": {
            "*": "counterTop.loaf1[&1].&",
            "test": "counterTop.loaf1[&1].test"
          }
        }
      }
    }
  },
  {
    "operation": "default",
    "spec": {
      "testMap": {
        "nio": "0",
        "io": "1"
      }
    }
  },
  {
    "operation": "modify-overwrite-beta",
    "spec": {
      "counterTop": {
        "loaf1": {
          "*": {
            "test": "=lookup(@(2,testMap),@(1,test))"
          }
        }
      }
    }
  },
  {
    "operation": "remove",
    "spec": {
      "testMap": ""
    }
  }
]

规则扩展方式：需要新增替换规则时，直接在testMap节点下添加对应的键值对即可，无需修改其他逻辑。

方案2：使用ExecuteScript处理器（Groovy实现，更易维护） #

处理器语言选择Groovy，填入如下脚本：

import groovy.json.JsonSlurper
import groovy.json.JsonBuilder

def flowFile = session.get()
if (!flowFile) return

try {
    // 替换规则配置，新增规则直接修改此处即可
    def testMapping = ["nio": "0", "io": "1"]
    
    def content = session.read(flowFile).withStream { stream ->
        new JsonSlurper().parse(stream)
    }
    
    content.counterTop.loaf1.each { item ->
        if (item.containsKey("test") && testMapping.containsKey(item.test)) {
            item.test = testMapping[item.test]
        }
    }
    
    flowFile = session.write(flowFile, { outStream ->
        outStream.write(new JsonBuilder(content).toPrettyString().getBytes("UTF-8"))
    } as OutputStreamCallback)
    
    session.transfer(flowFile, REL_SUCCESS)
} catch (Exception e) {
    session.transfer(flowFile, REL_FAILURE)
}

该方案语法更直观，适合后续需要叠加更复杂转换逻辑的场景。

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