如何在Apache NiFi中实现键值对映射？（含数据转换场景）

针对你提供的JSON结构转换需求，推荐两种可行的实现方案：

方案一：使用JoltTransformJSON处理器（无代码方式） #

Jolt是NiFi内置的JSON结构转换工具，适合处理这类键值映射场景。

配置步骤： #

1. 在NiFi画布中添加JoltTransformJSON处理器
2. 进入处理器配置界面，设置以下核心属性：
   • Jolt Specification：填入以下转换规则（自动取第一个columns数组作为键，与rows数据配对）

     [
       {
         "operation": "shift",
         "spec": {
           "results": {
             "rows": {
               "*": {
                 "@": "&3[&1].values",
                 "@(2,columns[0])": "&3[&1].keys"
               }
             }
           }
         }
       },
       {
         "operation": "shift",
         "spec": {
           "results": {
             "*": {
               "keys": {
                 "*": {
                   "@(2,values[&1])": "&4[&3].&"
                 }
               }
             }
           }
         }
       },
       {
         "operation": "remove",
         "spec": {
           "results": ""
         }
       }
     ]
     

   • Input JSON Path：保持默认值$（读取整个流文件的JSON内容）
   • Output JSON Path：保持默认值$（输出完整的转换结果）
3. 将输入流文件连接到该处理器的success关系，启动处理器即可得到期望的键值对数组输出。

方案二：使用ExecuteScript处理器（自定义代码方式） #

如果需要更灵活的逻辑控制，可以用Groovy脚本实现转换：

配置步骤： #

1. 在NiFi画布中添加ExecuteScript处理器
2. 配置处理器的Script Language为Groovy，并填入以下脚本：

   import groovy.json.JsonSlurper
   import groovy.json.JsonBuilder
   
   def flowFile = session.get()
   if (!flowFile) return
   
   flowFile = session.write(flowFile, { inputStream, outputStream ->
       def json = new JsonSlurper().parse(inputStream)
       // 提取首个columns数组作为键集合
       def keyList = json.results.columns[0]
       // 遍历rows生成键值对映射
       def transformedList = json.results.rows.collect { row ->
           def keyValueMap = [:]
           keyList.eachWithIndex { key, index ->
               keyValueMap[key] = row[index]
           }
           keyValueMap
       }
       // 写出转换后的JSON
       outputStream.write(new JsonBuilder(transformedList).toByteArray())
   } as StreamCallback)
   
   session.transfer(flowFile, REL_SUCCESS)
   

3. 连接输入流并启动处理器，即可完成转换。

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