如何使用NiFi处理器基于JSON键值对批量查找替换txt文件内容

我在NiFi中使用FetchSFTP读取txt文件，同时通过REST调用和JoltTransformJSON处理后得到了如下JSON格式的键值对：

[{
    "Key": "k2s2e2",
    "Value": "Ottawa"
}, {
    "Key": "60601",
    "Value": "Chicago"
}, {
    "Key": "",
    "Value": "London"
}]

需要将txt文件中所有与上述键匹配的内容替换为对应的值，示例如下：

• 原abc.txt内容：

000 apple stocks at k2s2e2 888
9000 samsung stocks at 60601 9990377
88 nokia devivces at 78889 790888071 hgj 7

• 期望输出：

000 apple stocks at Ottawa 888
9000 samsung stocks at Chicago 9990377
88 nokia devivces at 78889 790888071 hgj 7

你写的Groovy脚本存在几个核心问题：

1. 硬编码本地文件路径不符合NiFi的流处理逻辑，NiFi中所有文件内容都以FlowFile形式流转，不需要读写本地磁盘
2. 遍历每个键值对时重复读写文件，效率极低且容易出现内容覆盖错误
3. 使用replaceAll做替换，若键包含.、*、?等正则特殊字符会导致替换结果不符合预期
4. 最终写入FlowFile的是Writer对象的toString结果，不是替换后的文本内容

这里采用更符合NiFi使用习惯的逻辑：输入FlowFile为待替换的txt内容，之前生成的键值对JSON预先存入FlowFile的属性中（属性名设为replace_map，可通过UpdateAttribute等处理器实现），然后使用如下ExecuteGroovyScript代码完成替换：

import java.nio.charset.StandardCharsets
import groovy.json.JsonSlurper

def flowFile = session.get()
if(!flowFile) return

try {
    // 从属性中读取替换映射JSON并解析
    def replaceMap = [:]
    def mapJsonStr = flowFile.getAttribute('replace_map')
    def jsonSlurper = new JsonSlurper()
    def mapList = jsonSlurper.parseText(mapJsonStr)
    mapList.each {
        if(it.Key.trim() != "") {
            replaceMap[it.Key] = it.Value
        }
    }

    // 读取待替换的txt内容
    def content = flowFile.read().getText("UTF-8")

    // 批量替换所有匹配键，replace为全字符匹配不会解析正则
    replaceMap.each { key, value ->
        content = content.replace(key, value)
    }

    // 把替换后的内容写回流文件
    flowFile = session.write(flowFile, { out ->
        out.write(content.getBytes(StandardCharsets.UTF_8))
    } as OutputStreamCallback)

    session.transfer(flowFile, REL_SUCCESS)
} catch(Exception e) {
    log.error("替换内容失败", e)
    session.transfer(flowFile, REL_FAILURE)
}

如果你的输入FlowFile是键值对JSON，待替换的txt内容存在属性中，只需要调整读取JSON和文本的来源即可。

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