如何在NiFi中用Groovy/Python解码动态JSON数组的Base64值

需求说明 #

需要在NiFi中处理JSON数据，对动态长度数组A内每个对象的CC字段（Base64编码字符串）进行解码，其余字段保持不变。

Groovy 脚本实现（适配 ExecuteScript 处理器） #

将以下脚本复制到ExecuteScript处理器的Script Body中，语言选择Groovy：

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

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

flowFile = session.write(flowFile, { inputStream, outputStream ->
    // 读取并解析输入JSON
    def json = new JsonSlurper().parse(inputStream, StandardCharsets.UTF_8.name())
    
    // 遍历数组A的每个元素，解码CC字段
    if (json.A instanceof List) {
        json.A.each { item ->
            if (item.CC) {
                // Base64解码，转换为UTF-8字符串
                item.CC = new String(item.CC.decodeBase64(), StandardCharsets.UTF_8)
            }
        }
    }
    
    // 将处理后的JSON写入输出流
    outputStream.write(new JsonBuilder(json).toPrettyString().getBytes(StandardCharsets.UTF_8))
} as StreamCallback)

session.transfer(flowFile, REL_SUCCESS)

关键说明 #

• 用JsonSlurper和JsonBuilder完成JSON解析与序列化，保持代码简洁
• 动态判断数组A的存在性，兼容无该字段的场景
• 仅对非空的CC字段解码，避免空指针异常
• 输出格式化后的JSON，保持可读性

Python 脚本实现（适配 ExecuteScript 处理器） #

将以下脚本复制到ExecuteScript处理器的Script Body中，语言选择Python：

import json
import base64
from org.apache.commons.io import IOUtils
from java.nio.charset import StandardCharsets

flowFile = session.get()
if not flowFile:
    raise SystemExit()

# 读取输入流内容
input_content = IOUtils.toString(flowFile.read(), StandardCharsets.UTF_8.name())
json_data = json.loads(input_content)

# 处理数组A的CC字段解码
if isinstance(json_data.get('A'), list):
    for item in json_data['A']:
        if 'CC' in item and item['CC']:
            # Base64解码并转为UTF-8字符串
            decoded_bytes = base64.b64decode(item['CC'])
            item['CC'] = decoded_bytes.decode('utf-8')

# 写入处理后的JSON到输出流
flowFile = session.write(flowFile, lambda outputStream: outputStream.write(json.dumps(json_data, indent=2).encode('utf-8')))

session.transfer(flowFile, REL_SUCCESS)

关键说明 #

• 借助NiFi集成的IOUtils读取流内容，适配Python与Java的交互
• 使用Python标准库json和base64完成核心逻辑
• 增加字段存在性判断，提升脚本健壮性
• 输出格式化后的JSON，与输入格式保持一致

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