如何在Apache NiFi的JoltTransforms中实现对象优先级过滤逻辑？

需求说明 #

编写Jolt规范，对affectedEntities数组中的实体按优先级过滤：仅保留最高优先级的目标实体，丢弃其他低优先级实体及POLICY类型实体。

过滤规则 #

• 若存在application-component-node（对应entityType值为APPLICATION-Component-Node），仅保留该实体，丢弃application-component和application类型实体
• 若存在application-component（对应entityType值为APPLICATION-Component），仅保留该实体，丢弃application类型实体
• 若仅存在application（对应entityType值为APPLICATION），则保留该实体

实体优先级 #

• 优先级1：application-component-node
• 优先级2：application-component
• 优先级3：application

示例输入 #

[
  {
    "severity": "ERROR",
    "type": "POLICY_CONTINUES_CRITICAL",
    "affectedEntities": [
      {
        "entityType": "POLICY",
        "name": "Exception Per Minute",
        "entityId": 3683
      },
      {
        "entityType": "APPLICATION",
        "name": "Teashop",
        "entityId": 6026
      },
      {
        "entityType": "APPLICATION-Component",
        "name": "Teashop",
        "entityId": 602667
      }
    ],
    "subType": "OVERALL_APPLICATION",
    "id": 92876278,
    "triggeredEntity": null
  }
]

预期输出 #

[
  {
    "severity": "ERROR",
    "type": "POLICY_CONTINUES_CRITICAL",
    "affectedEntities": [
      {
        "entityType": "APPLICATION-Component",
        "name": "Teashop",
        "entityId": 602667
      }
    ],
    "subType": "OVERALL_APPLICATION",
    "id": 92876278,
    "triggeredEntity": null
  }
]

Jolt规范实现 #

[
  {
    "operation": "shift",
    "spec": {
      "*": {
        "affectedEntities": {
          "*": {
            "entityType": {
              "APPLICATION-Component-Node": {
                "@(2)": "&4.&3[#2]"
              },
              "APPLICATION-Component": {
                "@(2)": "&4.&3[#2]"
              },
              "APPLICATION": {
                "@(2)": "&4.&3[#2]"
              }
            }
          }
        },
        "*": "&1.&"
      }
    }
  },
  {
    "operation": "modify-overwrite-beta",
    "spec": {
      "*": {
        "affectedEntities": {
          "hasNode": "=size(@(1,APPLICATION-Component-Node))",
          "hasComponent": "=size(@(1,APPLICATION-Component))",
          "selected": "=ifElse(@(1,hasNode)>0, @(1,APPLICATION-Component-Node), ifElse(@(1,hasComponent)>0, @(1,APPLICATION-Component), @(1,APPLICATION)))"
        }
      }
    }
  },
  {
    "operation": "shift",
    "spec": {
      "*": {
        "affectedEntities": {
          "selected": "&2.&1"
        },
        "*": "&1.&"
      }
    }
  },
  {
    "operation": "remove",
    "spec": {
      "*": {
        "affectedEntities": {
          "hasNode": "",
          "hasComponent": "",
          "APPLICATION-Component-Node": "",
          "APPLICATION-Component": "",
          "APPLICATION": ""
        }
      }
    }
  }
]

规范解释 #

1. Shift操作（第一步）：仅提取affectedEntities中三个目标类型的实体并按entityType分组，忽略POLICY类型实体，其余顶层字段原样保留。
2. Modify操作（第二步）：通过计算各类型实体的数量判断是否存在，再用嵌套ifElse按优先级筛选出要保留的实体数组。
3. Shift操作（第三步）：将筛选后的实体数组替换回affectedEntities字段。
4. Remove操作（第四步）：清理过程中生成的临时分组字段，保证最终结构整洁。

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