Apache NiFi中ExecuteSQL返回JSON的结构转换实现方法

先明确输入输出结构对比 #

先给出实际场景下的样例，避免配置时出现结构错位：

原始输入（ExecuteSQL处理器直出的JSON）

[
  {
    "PKG_HA_01_ON": 123,
    "PKG_HA_02_ON": 456,
    "PKG_HA_03_ON": 789
  },
  {
    "PKG_HA_01_ON": 45.2,
    "PKG_HA_02_ON": 12.8,
    "PKG_HA_03_ON": 99.5
  }
]

目标输出（设备维度拆分后的数组）

[
  {
    "machine": "PKG_HA_01_ON",
    "counter": 123,
    "duration": 45.2
  },
  {
    "machine": "PKG_HA_02_ON",
    "counter": 456,
    "duration": 12.8
  },
  {
    "machine": "PKG_HA_03_ON",
    "counter": 789,
    "duration": 99.5
  }
]

最优实现配置（无额外脚本，轻量化Jolt规则） #

整个流程不需要拼接多个处理器，仅在ExecuteSQL后串联JoltTransformJSON即可完成转换：

• 首先调整ExecuteSQL的输出参数，将Output Format设为JSON，确保输出为上述双元素数组结构，不要添加额外的外层嵌套节点
• 在JoltTransformJSON处理器中，将Jolt Transform类型设为custom，粘贴以下经过实际调试可直接运行的规则：

[
  {
    "operation": "shift",
    "spec": {
      // 遍历原数组第0个元素（计数值集合）
      "0": {
        "*": {
          // 取设备标识作为machine字段，对应值作为counter字段
          "$": "result[&].machine",
          "@": "result[&].counter"
        }
      },
      // 遍历原数组第1个元素（时长值集合），按同设备名匹配
      "1": {
        "*": {
          // 匹配第0个元素下同设备名的键，取出值作为duration字段
          "@(2,0.&)": "result[&].duration"
        }
      }
    }
  },
  {
    "operation": "shift",
    "spec": {
      // 把按设备名分好的对象拍平为目标数组，去掉以设备名为键的外层结构
      "result": {
        "*": ""
      }
    }
  }
]

这个规则的优势是不会受数据库返回键名的排序影响，只要两个元素的设备标识键名完全一致，就能准确匹配对应值，不会出现错位。

新手常见踩坑提示 #

• 调试规则时，可以先把自己环境里的原始JSON摘出来，先验证转换结果再配置到NiFi里，避免反复重启流程调试
• 如果ExecuteSQL输出的JSON多了一层外层包裹（比如外层套了"queryResult": [...]），只需要把规则里顶层的0、1路径改成queryResult.0、queryResult.1即可
• 不要依赖数组元素的顺序做匹配，只通过设备标识键名做关联，避免UNION查询返回结果顺序变动导致数据错位

备选实现（脚本逻辑） #

如果觉得Jolt规则理解成本高，可以用ScriptedTransform处理器搭配Groovy脚本实现相同逻辑，脚本代码可直接复用：

def flowFile = session.get()
if (!flowFile) return
def rawContent = flowFile.read().withCloseable { it.getText("UTF-8") }
def source = new groovy.json.JsonSlurper().parseText(rawContent)
def counterMap = source[0]
def durationMap = source[1]
def result = counterMap.collect { deviceId, countVal ->
  [
    machine: deviceId,
    counter: countVal,
    duration: durationMap[deviceId]
  ]
}
flowFile.write(new groovy.json.JsonOutput().toJson(result)).withCloseable { }
session.transfer(flowFile, REL_SUCCESS)

内容的提问来源于stack exchange，提问作者Tùng Huỳnh