如何通过Apache Nifi JoltTransform处理器动态过滤Flowfile Payload字段

需求说明 #

根据输入JSON中keys数组指定的字段名，从payload对象中筛选出对应字段，保留id字段并移除keys数组。

输入JSON #

{"id": 1,"payload": {"code": "123-ABC2","code3": "123-ABC3","code2": "123-ABC"},"keys": [{"key": "code"},{"key": "code2"}]}

目标输出JSON #

{"id": 1,"payload": {"code": "123-ABC2","code2": "123-ABC"}}

Jolt转换规范 #

将以下JSON配置输入到JoltTransform处理器的"Jolt Specification"属性中：

[
  {
    "operation": "shift",
    "spec": {
      "id": "id",
      "payload": {
        "*": {
          "@(2,keys[*].key)": {
            "$": {
              "@2": "payload.&"
            }
          }
        }
      }
    }
  }
]

逻辑解释 #

1. 保留id字段：通过"id": "id"直接将原id字段映射到输出的同名字段。
2. 动态筛选payload字段：
   • payload.*遍历payload下的所有字段
   • @(2,keys[*].key)向上两层获取keys数组中所有key值，形成字段名集合
   • $代表当前遍历到的payload字段名，判断该名称是否存在于上述集合中
   • 若匹配成功，通过@2获取该字段的原始值，映射到输出的payload.&中（&复用匹配的字段名）
3. 自动移除keys数组：由于spec中未对keys字段做任何映射，转换后该字段会被自动丢弃。

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