如何使用NiFi筛选JSON文件中指定属性的条目？

需求背景 #

处理包含eventType、eventTime、country、resourceType字段的JSON对象数组，需要筛选出country值为ES的条目，最终得到仅包含符合条件条目的JSON文件。目前已能通过EvaluateJsonPath处理器提取JSON属性，希望完成后续筛选逻辑。

原始JSON内容 #

[
  {
    "eventType": "UPDATE",
    "eventTime": "2021-12-14T12:34:56.789012Z",
    "country": "ES",
    "resourceType": "Party"
  },
  {
    "eventType": "UPDATE",
    "eventTime": "2021-12-14T12:34:56.789012Z",
    "country": "ES",
    "resourceType": "Party"
  },
  {
    "eventType": "UPDATE",
    "eventTime": "2021-12-14T12:34:56.789012Z",
    "country": "FR",
    "resourceType": "Party"
  },
  {
    "eventType": "UPDATE",
    "eventTime": "2021-12-14T12:34:56.789012Z",
    "country": "FR",
    "resourceType": "Party"
  }
]

期望结果 #

[
  {
    "eventType": "UPDATE",
    "eventTime": "2021-12-14T12:34:56.789012Z",
    "country": "ES",
    "resourceType": "Party"
  },
  {
    "eventType": "UPDATE",
    "eventTime": "2021-12-14T12:34:56.789012Z",
    "country": "ES",
    "resourceType": "Party"
  }
]

实现步骤 #

注意：UpdateAttribute处理器仅用于修改流文件属性，无法直接筛选JSON条目或修改JSON内容，需结合以下处理器完成需求：

1. SplitJson处理器

   • 配置Json Path Expression为$[*]，将JSON数组拆分为单个JSON对象，每个对象对应一个独立流文件。

2. EvaluateJsonPath处理器

   • 添加属性country，设置Json Path为$.country，返回类型选择Scalar，将每个对象的country值提取为流文件属性。

3. RouteOnAttribute处理器

   • 添加路由规则isES，表达式为${country:equals('ES')}，匹配该规则的流文件发送到matched关系，不匹配的发送到unmatched关系（可后续丢弃或处理）。

4. MergeContent处理器

   • 合并matched关系中的流文件，配置：
     • Merge Strategy：Bin-Packing Algorithm
     • Delimiter Strategy：Text
     • Header：[
     • Footer：]
     • Demarcator：,\n
   • 确保合并后输出合法的JSON数组。

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