如何在Apache NiFi中使用Jolt规范拆分嵌套JSON？

输入JSON #

{
  "VulnId": 90089,
  "source": "WERUA",
  "sourceId": "WERT38173",
  "StateDescription": "Not Being Used",
  "Status": true,
  "DurationInDays": "69",
  "Assessments": [
    {"id": 357978,"name": "2023-02-05T10:00:15.620","date": "2023-02-05T10:00:15","type": "System"},
    {"id": 370904,"name": "2023-03-14T10:00:05.872 ","date": "2023-03-14T10:00:05","type": "System"}
  ],
  "Tags": [
    {"id": 47954778,"name": "RS_Co","category": "Custom","created": "2023-06-19T16:59:37","updated": "2023-06-19T16:59:37"},
    {"id": 479547767,"name": "RS_Co","category": "Custom","created": "2023-06-19T16:59:37","updated": "2023-06-19T16:59:37"}
  ],
  "workflowDistribution_latestSystemWorkflows": [
    {"generatedId": "SY00989","type": "SYSTEM","acronym": "SYS","state": "Approved"},
    {"generatedId": "SY8989","type": "SYSTEM","acronym": "SYS","state": "Approved"}
  ],
  "firstAssignedOnDate": null
}

预期输出JSON #

[
  {
    "VulnId": 90089,
    "source": "WERUA",
    "sourceId": "WERT38173",
    "StateDescription": "Not Being Used",
    "Status": true,
    "DurationInDays": "69",
    "Assessments": {"id": 357978,"name": "2023-02-05T10:00:15.620 ","date": "2023-02-05T10:00:15","type": "System"},
    "Tags": {"id": 47954778,"name": "RS_Co","category": "Custom","created": "2023-06-19T16:59:37","updated": "2023-06-19T16:59:37"},
    "workflowDistribution_latestSystemWorkflows": {"generatedId": "SY00989","type": "SYSTEM","acronym": "SYS","state": "Approved"},
    "firstAssignedOnDate": null
  },
  {
    "VulnId": 90089,
    "source": "WERUA",
    "sourceId": "WERT38173",
    "StateDescription": "Not Being Used",
    "Status": true,
    "DurationInDays": "69",
    "Assessments": {"id": 370904,"name": "2023-03-14T10:00:05.872 ","date": "2023-03-14T10:00:05","type": "System"},
    "Tags": {"id": 479547767,"name": "RS_Co","category": "Custom","created": "2023-06-19T16:59:37","updated": "2023-06-19T16:59:37"},
    "workflowDistribution_latestSystemWorkflows": {"generatedId": "SY8989","type": "SYSTEM","acronym": "SYS","state": "Approved"},
    "firstAssignedOnDate": null
  }
]

Jolt规范 #

[
  {
    "operation": "shift",
    "spec": {
      "VulnId": "[&1].VulnId",
      "source": "[&1].source",
      "sourceId": "[&1].sourceId",
      "StateDescription": "[&1].StateDescription",
      "Status": "[&1].Status",
      "DurationInDays": "[&1].DurationInDays",
      "firstAssignedOnDate": "[&1].firstAssignedOnDate",
      "Assessments": {
        "*": {
          "@": "[&1].Assessments"
        }
      },
      "Tags": {
        "*": {
          "@": "[&1].Tags"
        }
      },
      "workflowDistribution_latestSystemWorkflows": {
        "*": {
          "@": "[&1].workflowDistribution_latestSystemWorkflows"
        }
      }
    }
  },
  {
    "operation": "default",
    "spec": {
      "*": {
        "VulnId": "",
        "source": "",
        "sourceId": "",
        "StateDescription": "",
        "Status": "",
        "DurationInDays": "",
        "firstAssignedOnDate": null,
        "Assessments": {},
        "Tags": {},
        "workflowDistribution_latestSystemWorkflows": {}
      }
    }
  },
  {
    "operation": "remove",
    "spec": {
      "*": {
        "VulnId": ["", null],
        "source": ["", null],
        "sourceId": ["", null],
        "StateDescription": ["", null],
        "Status": ["", null],
        "DurationInDays": ["", null],
        "Assessments": [{}],
        "Tags": [{}],
        "workflowDistribution_latestSystemWorkflows": [{}]
      }
    }
  }
]

规范说明 #

1. Shift操作：核心转换步骤，将所有顶级非数组字段复制到每个索引对应的节点，同时把三个数组的元素按索引一一映射到对应节点下，实现按索引展开对象的效果。
2. Default操作：为每个节点的所有字段设置默认值，避免因数组长度不一致导致字段缺失。
3. Remove操作：可选步骤，用于清理空值或空对象字段，保证输出结构整洁。

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