如何使用JOLT Transform将NiFi中的MapRecord转为扁平化JSON

因为输入中issues、address字段为带MapRecord[{}]标识的非标准JSON字符串，无法直接被JOLT识别解析，需要先做格式预处理，再执行JOLT转换，完整步骤如下：

步骤1：预处理MapRecord字符串为标准JSON结构 #

使用NiFi ReplaceText 处理器配置两次替换规则，将非标准格式转成标准JSON对象：

• 第一次替换：把键值对的=转为标准JSON的键值格式
  • 搜索正则：(\w+)=([^,}\]]+)
  • 替换值："$1":"$2"
  • 替换模式：替换所有出现
• 第二次替换：剥掉外层MapRecord[]包裹
  • 搜索正则："MapRecord\[({.*?})\]"
  • 替换值：$1
  • 替换模式：替换所有出现

预处理完成后得到的中间输入结构如下：

{
   "maxResults": 150,
   "total": 89,
   "issues": {"issueId":1, "firstName":"Jack", "lastName":"Smith"},
   "address": {"addressId":1, "street":"Mockingbird Lane", "town":"Timbuktoo"}
}

步骤2：配置JOLT Transform执行扁平化 #

使用JOLT的shift转换类型，转换规则如下：

[
  {
    "operation": "shift",
    "spec": {
      // 保留顶层原有字段
      "maxResults": "maxResults",
      "total": "total",
      // 提取issues下所有子字段到顶层
      "issues": {
        "*": "&"
      },
      // 提取address下所有子字段到顶层
      "address": {
        "*": "&"
      }
    }
  },
  // 可选：移除不需要的冗余字段
  {
    "operation": "remove",
    "spec": {
      "issueId": "",
      "addressId": ""
    }
  }
]

执行后即可得到目标扁平化输出结果。

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