如何使用NiFi将CSV文件转换为嵌套格式JSON？

示例CSV数据 #

目标嵌套JSON格式 #

[
{"Agent":"A1",
  "loc_id":"L1",
   "Company":{ 
        "Hyundai":[ {"Verna":[
                              {
                               "2018":[]
                              }
                             ]
                    },
                    {"Creta":[
                              {
                                "2015":[]
                              }
                             ]
                    },
                    {"Aura":[
                              {
                               "2022":[]
                              }
                             ]
                    }
                ]
             }
},
{"Agent":"A2",
 "loc_id":"L1",
 "Company":{
             "Toyota":[
                      {
                       "Fortuner":[
                                  {
                                    "2020":[]
                                  }
                                  ]
                      }
                      ]
           }
}
]

具体实现步骤 #

1. 将CSV转换为单条JSON记录

使用ConvertRecord处理器：

• Reader：选择CSVReader，配置表头为Agent,Location,Company,Model1,Sub-Model1，确保字段映射匹配。
• Writer：选择JSONRecordSetWriter，设置输出格式为JSON Lines，让每条CSV行对应一个独立的JSON对象。

2. 按Agent和Location分组合并记录

使用MergeContent处理器：

• 设置Merge Strategy为Attribute Value，Attribute Name填写表达式${Agent}-${Location}，确保同Agent同Location的JSON记录被合并为一个数组。
• 设置Delimiter Strategy为Text，Header设为[，Footer设为]，Delimiter设为,，生成按分组聚合的JSON数组。

3. 使用JoltTransformJSON实现嵌套转换

这是核心步骤，在JoltTransformJSON的Jolt Specification中输入以下转换规则：

[
  {
    "operation": "shift",
    "spec": {
      "*": {
        "Agent": "[#2].Agent",
        "Location": "[#2].loc_id",
        "Company": "[#2].Company.&",
        "Model1": "[#2].Company.${Company}[#3].&",
        "Sub-Model1": "[#2].Company.${Company}[#3].${Model1}[#4].&[]"
      }
    }
  },
  {
    "operation": "modify-overwrite-beta",
    "spec": {
      "*": {
        "Company": {
          "*": {
            "*": {
              "*": {
                "*": "[]"
              }
            }
          }
        }
      }
    }
  },
  {
    "operation": "cardinality",
    "spec": {
      "*": {
        "Agent": "ONE",
        "loc_id": "ONE",
        "Company": {
          "*": "MANY"
        }
      }
    }
  }
]

4. 输出最终JSON

使用PutFile处理器将转换后的JSON写入指定存储路径，完成整个转换流程。

说明 #

• Jolt规范中，shift操作负责将扁平字段映射到目标嵌套层级；modify-overwrite-beta操作将Sub-Model的值替换为空数组；cardinality操作确保Agent和loc_id只保留唯一值，Company下的数组保留所有元素。
• 如果CSV有更多嵌套层级，可调整Jolt规范中的路径表达式适配需求。

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