Apache NiFi中Java实现JSON字段替换与批量结果生成

一、ExecuteScript处理器的Java实现 #

直接在ExecuteScript处理器中使用以下Java代码，替换原有脚本：

import org.apache.nifi.processor.io.StreamCallback;
import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.fasterxml.jackson.databind.node.ArrayNode;
import com.fasterxml.jackson.databind.node.ObjectNode;

import java.io.InputStream;
import java.io.OutputStream;

// 从属性获取原始数据
String apiResponse = context.getProperty("Response").getValue();
String templateJson = context.getProperty("jsonData").getValue();

ObjectMapper objectMapper = new ObjectMapper();
JsonNode responseRoot = objectMapper.readTree(apiResponse);
JsonNode templateRoot = objectMapper.readTree(templateJson);

ArrayNode resultArray = objectMapper.createArrayNode();

// 遍历API返回的每个子数组，提取ID并替换模板
for (JsonNode subArray : responseRoot) {
    if (subArray.isArray() && subArray.size() > 0) {
        String targetId = subArray.get(0).asText();
        // 深拷贝模板对象，避免修改原数据
        ObjectNode newJsonObj = (ObjectNode) templateRoot.deepCopy();
        newJsonObj.put("Emp_Id", targetId);
        resultArray.add(newJsonObj);
    }
}

// 将结果写入流文件内容
session.write(flowFile, new StreamCallback() {
    @Override
    public void process(InputStream in, OutputStream out) {
        out.write(objectMapper.writeValueAsBytes(resultArray));
    }
});

配置说明 #

• 处理器的Script Language选择Java
• 确保Response和jsonData属性名与ExtractText处理器输出的属性完全一致
• 依赖的Jackson库是NiFi默认自带的，无需额外添加

二、无脚本的纯NiFi处理器组合方案 #

如果想避开脚本执行的问题，可通过以下处理器组合实现需求：

1. SplitJson

   • 作用：拆分API返回的Response数据中的子数组
   • 配置：Json Path Expression设为$[*]，将每个子数组拆分为独立的流文件

2. ExtractText

   • 作用：从拆分后的子数组中提取第一个元素（ID）
   • 配置：添加属性Emp_Id，正则表达式根据ID格式调整，例如：
     • 若ID是数字：^\[(\d+),.*\]$，替换值设为$1
     • 若ID是字符串：^\["([^"]+)",.*\]$，替换值设为$1

3. AttributeToJson

   • 作用：将jsonData属性转换为流文件内容
   • 配置：Attributes to JSON选择Specified Attributes，指定jsonData属性

4. ReplaceText

   • 作用：替换JSON内容中的"Emp_Id":"All"为提取到的ID
   • 配置：
     • Search Value："Emp_Id":\s*"All"
     • Replacement Value："Emp_Id": "${Emp_Id}"
     • Replacement Strategy：Replace All

5. MergeContent

   • 作用：将所有替换后的JSON对象合并为一个数组
   • 配置：
     • Merge Strategy：Bin-Packing Algorithm
     • Delimiter Strategy：Text
     • Header：[
     • Footer：]
     • Demarcator：,

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