基于temp字段值过滤JSON中items数组的Jolt/NiFi问题

需求说明 #

需要根据temp的值（格式为|分隔的字符串），过滤JSON中data[0].product.items数组，只保留item_number匹配temp中值的元素，同时保留原JSON的其他结构。

输入JSON #

{"temp": "1010328|1010329", "data": [{"product": {"items": [{"item_number": "1010328", "attributes": [{"id": "myId01", "type": "int"}]}, {"item_number": "1010329", "attributes": [{"id": "myId02", "type": "int"}]}, {"item_number": "1010330", "attributes": [{"id": "myId03", "type": "int"}]}]}}]}

期望输出JSON #

{"temp": "1010328|1010329", "data": [{"product": {"items": [{"item_number": "1010328", "attributes": [{"id": "myId01", "type": "int"}]}, {"item_number": "1010329", "attributes": [{"id": "myId02", "type": "int"}]}]}}]}

解决方案 #

由于Jolt的filter-tasks不支持直接引用NiFi变量，分两种场景处理：

场景1：用JSON内的temp字段过滤 #

使用Jolt的多步操作实现：

1. 先拆分temp字符串为数组；
2. 基于数组匹配过滤items元素；
3. 清理临时字段。

对应的Jolt规格：

[
  {
    "operation": "modify-overwrite-beta",
    "spec": {
      "temp_list": "=split('|', @(1,temp))"
    }
  },
  {
    "operation": "shift",
    "spec": {
      "temp": "temp",
      "data": {
        "*": {
          "product": {
            "items": {
              "*": {
                "item_number": {
                  "@(2,temp_list)": {
                    "@(3)": "data[&4].product.items[]"
                  }
                }
              }
            }
          }
        }
      }
    }
  },
  {
    "operation": "remove",
    "spec": {
      "temp_list": ""
    }
  }
]

场景2：用NiFi变量temp过滤 #

如果temp是NiFi变量，需要先将变量注入JSON，再执行过滤：

• 用NiFi的ReplaceText或UpdateRecord处理器，把NiFi变量${temp}插入到输入JSON的顶层，比如添加"temp": "${temp}"字段；
• 再用上述Jolt规格完成过滤；
• 若不需要保留temp字段，可在最后一步的remove操作中删除它。

内容的提问来源于stack exchange，提问作者Katja Bürger