如何在NiFi处理器中拆分逗号分隔字段并拼接字段前3个词？

需求说明 #

需在NiFi处理器中拆分逗号分隔的字段，并将字段1与字段2的前3个词进行拼接。

示例：
输入内容：2022-09-05T00:00:10,677 abc.1 ,
拆分并拼接后的结果：2022-09-05T00:00:10:677,abc.1,

实现方案 #

方法一：使用UpdateRecord处理器（推荐） #

1. 配置CSV记录读取器
   • 创建CSVReader控制器服务，设置分隔符为逗号，开启忽略尾部空格，按需配置跳过空字段。
2. 配置CSV记录写入器
   • 创建CSVRecordSetWriter控制器服务，设置分隔符为逗号，匹配输出格式要求。
3. 配置UpdateRecord处理器
   • 指定记录读取器为上述CSVReader，记录写入器为上述CSVRecordSetWriter。
   • 添加字段映射规则：
     • 目标拼接字段（如combined_field）使用NiFi表达式语言实现逻辑：

       ${field1}:${field2:split('\\s+'):get(0)},${field2:split('\\s+'):get(1)}
       

       注：若字段2的有效词不足3个，表达式会自动取现有部分；需严格取前3个可添加判断逻辑扩展。
     • 按需保留或调整其他输出字段。

方法二：使用ExecuteScript处理器（灵活定制） #

选择Groovy脚本语言，编写脚本实现字段拆分与拼接：

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

flowFile = session.write(flowFile, { inputStream, outputStream ->
    def content = inputStream.text.trim()
    def fields = content.split(',')
    if (fields.length >= 2) {
        def field1 = fields[0].trim()
        def field2Parts = fields[1].trim().split(/\s+/)
        // 取字段2的前3个词，不足则取全部
        def targetParts = field2Parts.take(3)
        // 按示例格式拼接输出
        def result = "${field1}:${targetParts.join(',')},"
        outputStream.write(result.getBytes('UTF-8'))
    } else {
        // 处理格式异常的情况，直接输出原内容或自定义逻辑
        outputStream.write(content.getBytes('UTF-8'))
    }
} as StreamCallback)

session.transfer(flowFile, REL_SUCCESS)

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