Apache NiFi 2.0.0-M4：如何将/message值转为JSON主体？

下面提供两种实用的实现方法，精准匹配你的需求——丢弃原始JSON的其他字段，仅保留message字段解析后的JSON作为新的流文件内容：

方法一：使用GroovyScript处理器 #

通过Groovy脚本直接完成解析与替换，灵活性拉满：

1. 添加GroovyScript处理器，替换默认脚本为以下内容：

import groovy.json.JsonSlurper
import groovy.json.JsonBuilder

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

try {
    // 读取原始JSON内容
    def originalContent = flowFile.read().text
    def jsonSlurper = new JsonSlurper()
    def rootJson = jsonSlurper.parseText(originalContent)
    
    // 解析message字段中的转义JSON字符串
    def messageContent = rootJson.message
    def targetJson = jsonSlurper.parseText(messageContent)
    
    // 生成格式化的目标JSON并写入流文件
    def newContent = new JsonBuilder(targetJson).toPrettyString()
    flowFile = session.write(flowFile, { out ->
        out.write(newContent.getBytes('UTF-8'))
    } as OutputStreamCallback)
    
    session.transfer(flowFile, REL_SUCCESS)
} catch (Exception e) {
    log.error("处理失败：${e.getMessage()}", e)
    session.transfer(flowFile, REL_FAILURE)
}

2. 配置处理器的REL_SUCCESS和REL_FAILURE连接到后续组件即可。

方法二：使用UpdateRecord处理器（声明式处理） #

利用NiFi原生Record API完成操作，无需编写脚本：

1. 先创建两个控制器服务：

   • JSONTreeReader：无需额外配置，默认即可解析原始JSON结构。
   • JSONRecordSetWriter：
     • 设置Schema Write Strategy为No Schema；
     • 可选勾选Pretty Print JSON让输出更易读。

2. 配置UpdateRecord处理器：

   • 选择刚才创建的JSONTreeReader作为Record Reader；
   • 选择刚才创建的JSONRecordSetWriter作为Record Writer；
   • 添加一个自定义属性：
     • 属性名：/（代表根节点，即整个JSON主体）
     • 属性值：${field.value:jsonPath('$.message'):jsonParse()}
       这个表达式的作用是：提取原始JSON的message字段值，将其从转义字符串解析为JSON对象，然后替换整个根节点内容。

3. 连接处理器的REL_SUCCESS和REL_FAILURE到对应组件。

注意事项 #

• 如果存在message字段为空或格式非法的情况，建议搭配RouteOnAttribute或HandleError处理器处理异常流。

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