基于NiFi的Jolt转换实现JSON数组按Key聚合并入库

需求概述 #

• 对输入的JSON数组按Key字段分组聚合
• 把同一Key下的所有itemNm整理成嵌套的Items数组（每个元素是带itemNm的对象）
• 聚合完成后，以Key为标识将数据存入数据库

输入JSON #

[
  {
    "Key": "991641500~2167767723",
    "itemNm": "2067875000"
  },
  {
    "Key": "991641500~2167767724",
    "itemNm": "2067875085"
  },
  {
    "Key": "991641500~2167767723",
    "itemNm": "2067875063"
  },
  {
    "Key": "991641500~2167767724",
    "itemNm": "2067875004"
  }
]

修正后期望输出JSON #

注：原需求中的期望输出存在itemNm与Key的对应错误，以下是与输入数据匹配的正确输出：

[
  {
    "Key": "991641500~2167767723",
    "Items": [
      {
        "itemNm": "2067875000"
      },
      {
        "itemNm": "2067875063"
      }
    ]
  },
  {
    "Key": "991641500~2167767724",
    "Items": [
      {
        "itemNm": "2067875085"
      },
      {
        "itemNm": "2067875004"
      }
    ]
  }
]

Jolt转换规则（Spec） #

在NiFi的JoltTransformJSON处理器中，选择Shift操作类型，使用以下规则：

[
  {
    "operation": "shift",
    "spec": {
      "*": {
        "@Key": "@Key",
        "@itemNm": "@Key.Items[#2].itemNm"
      }
    }
  },
  {
    "operation": "shift",
    "spec": {
      "*": {
        "Key": "[#2].Key",
        "Items": "[#2].Items"
      }
    }
  }
]

规则拆解 #

1. 第一步：分组收集数据
   遍历输入数组的每一条元素，用当前元素的Key值作为临时分组标识，把对应的itemNm收集到该分组的Items数组里。@Key用来引用当前元素的Key值作为临时结构的键，@itemNm引用itemNm的值，[#2]负责给同组的itemNm自动分配数组索引，生成有序的数组元素。

2. 第二步：转换为目标数组格式
   第一步生成的是{ "具体Key值": { "Key": "...", "Items": [...] } }的键值对结构，这一步把它转换成我们需要的数组形式，每个分组对象依次成为输出数组的元素。

NiFi后续入库处理 #

聚合完成后，用PutDatabaseRecord处理器把数据存入数据库：

• 配置好数据库连接池、目标表信息
• 选择JsonTreeReader作为Record Reader，根据数据库类型选对应的Record Writer（比如AvroRecordWriter）
• 确保Key字段与数据库表的对应字段映射正确，以Key作为标识完成数据写入

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