使用NiFi Jolt基于Aud_id合并API JSON响应并按条件过滤

需求规则 #

• 基于两个API响应中共有的唯一标识Aud_id合并JSON数据
• 仅保留在两个API响应中均存在Aud_id的条目
• 仅当API-1中的endDate满足距离当前日期小于30天的条件时，才执行合并操作

前置处理（NiFi流程） #

由于Jolt本身不支持日期计算，需先通过NiFi处理器完成日期差值计算：

1. 用InvokeHTTP分别调用两个API，获取响应内容
2. 用UpdateAttribute生成当前日期（格式yyyy-MM-dd），设置属性currentDate为${now():format('yyyy-MM-dd')}
3. 用ExecuteScript（Groovy脚本）为API-1的每个条目添加daysUntilEnd字段，计算endDate与当前日期的天数差：

def api1Json = new groovy.json.JsonSlurper().parseText(flowFile.content)
def currentDate = Date.parse("yyyy-MM-dd", attribute.currentDate)
api1Json.each { item ->
    def endDate = Date.parse("yyyy-MM-dd", item.endDate)
    def days = (endDate - currentDate).days
    item.daysUntilEnd = days
}
flowFile.content = new groovy.json.JsonBuilder(api1Json).toByteArray()
return [flowFile]

4. 用MergeContent将预处理后的API-1响应和API-2响应合并为如下结构的JSON：

{
  "api1": [...],
  "api2": [...]
}

Jolt转换规范 #

以下Jolt规范实现合并与过滤逻辑：

[
  // 第一步：将API-2数组转为以Aud_id为键的Map，快速匹配
  {
    "operation": "shift",
    "spec": {
      "api2": {
        "*": {
          "Aud_id": "&",
          "Aud_count": "&.Aud_count"
        }
      },
      "api1": "api1"
    }
  },
  // 第二步：遍历API-1条目，过滤符合日期条件的并合并API-2字段
  {
    "operation": "shift",
    "spec": {
      "api1": {
        "*": {
          "daysUntilEnd": {
            "<30": {
              "@2": "[&3]", // 保留API-1原有字段
              "@(3,api2.@(2,Aud_id).Aud_count)": "[&3].Aud_count" // 匹配对应Aud_id的Aud_count
            }
          }
        }
      }
    }
  },
  // 第三步：移除临时计算字段daysUntilEnd
  {
    "operation": "remove",
    "spec": {
      "*": {
        "daysUntilEnd": ""
      }
    }
  }
]

测试示例 #

预处理后的输入（合并后的JSON） #

{
  "api1": [
    {
      "Aud_id": "7f9a6541",
      "aud_name": "Holders_Current",
      "startDate": "2025-05-09",
      "endDate": "2026-01-31",
      "daysUntilEnd": 206
    },
    {
      "Aud_id": "463e9d0c",
      "aud_name": "Expiring_Current",
      "startDate": "2025-05-15",
      "endDate": "2025-07-31",
      "daysUntilEnd": 77
    },
    {
      "Aud_id": "4c973b25",
      "aud_name": "Reminder_Current",
      "startDate": "2025-05-15",
      "endDate": "2025-07-31",
      "daysUntilEnd": 77
    },
    {
      "Aud_id": "6c915c60",
      "aud_name": "Inactive_14-60days",
      "startDate": "2025-05-09",
      "endDate": "2025-12-31",
      "daysUntilEnd": 236
    },
    {
      "Aud_id": "87d32b36",
      "aud_name": "Inactive_14-60days",
      "startDate": "2024-12-02",
      "endDate": "2025-01-04",
      "daysUntilEnd": 3
    }
  ],
  "api2": [
    {
      "Aud_id": "7f9a6541",
      "Aud_count": "42526"
    },
    {
      "Aud_id": "463e9d0c",
      "Aud_count": "63737"
    },
    {
      "Aud_id": "4c973b25",
      "Aud_count": "728282"
    },
    {
      "Aud_id": "6c915c60",
      "Aud_count": "282"
    },
    {
      "Aud_id": "87d32b36",
      "Aud_count": "27278"
    }
  ]
}

输出结果（仅保留符合日期条件的条目） #

[
  {
    "Aud_id": "87d32b36",
    "aud_name": "Inactive_14-60days",
    "startDate": "2024-12-02",
    "endDate": "2025-01-04",
    "Aud_count": "27278"
  }
]

调整说明 #

• 若“endDate小于30天”指的是endDate与startDate的间隔小于30天，可修改前置脚本中的日期差值计算逻辑，改为计算两个日期的间隔
• 若两个API响应中存在仅单边有的Aud_id，Jolt会自动忽略无法匹配的条目，无需额外处理

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