NiFi Python脚本解析JSON设置文件名报错求助

你遇到的putAttribute(): 3rd arg can't be coerced to String异常，核心原因是你直接把FlowFileParser类的实例传给了putAttribute的第三个参数——而这个方法要求传入的是字符串类型的属性值，不是类实例。

下面是修正后的代码，分两种场景适配你的需求：

场景1：只修改文件名，保留原FlowFile的JSON内容 #

如果不需要改动FlowFile的原始内容，只是提取priority值作为新文件名，可以用这段代码：

import json
import java.io
from org.apache.commons.io import IOUtils
from java.nio.charset import StandardCharsets
from org.apache.nifi.processor.io import StreamCallback

class FlowFileParser(StreamCallback):
    def __init__(self):
        self.priority_value = None  # 用实例属性存储提取到的优先级值

    def process(self, inputStream, outputStream):
        # 读取FlowFile的JSON内容
        text = IOUtils.toString(inputStream, StandardCharsets.UTF_8)
        obj = json.loads(text)
        # 提取priority值保存到实例属性
        self.priority_value = obj['priority']
        # 将原内容写回输出流，保留原始JSON
        outputStream.write(bytearray(text.encode('utf-8')))

flowFile = session.get()
if flowFile != None:
    parser = FlowFileParser()
    # 先执行流处理，触发process方法提取priority值
    flowFile = session.write(flowFile, parser)
    # 用提取到的字符串值设置filename属性
    flowFile = session.putAttribute(flowFile, "filename", parser.priority_value)
    session.transfer(flowFile, REL_SUCCESS)

场景2：修改文件名的同时，替换FlowFile内容为priority值 #

如果需要把FlowFile的内容也改成priority的值（匹配你原代码的逻辑），可以用这段代码：

import json
import java.io
from org.apache.commons.io import IOUtils
from java.nio.charset import StandardCharsets
from org.apache.nifi.processor.io import StreamCallback

class FlowFileParser(StreamCallback):
    def __init__(self):
        self.priority_value = None

    def process(self, inputStream, outputStream):
        text = IOUtils.toString(inputStream, StandardCharsets.UTF_8)
        obj = json.loads(text)
        self.priority_value = obj['priority']
        # 将priority值作为新内容写回输出流
        outputStream.write(bytearray(self.priority_value.encode('utf-8')))

flowFile = session.get()
if flowFile != None:
    parser = FlowFileParser()
    flowFile = session.write(flowFile, parser)
    flowFile = session.putAttribute(flowFile, "filename", parser.priority_value)
    session.transfer(flowFile, REL_SUCCESS)

关键注意点 #

• 必须先调用session.write()触发process方法执行，这样才能从JSON中提取到priority值并存储到实例属性中
• 如果你的JSON里priority是数字类型（比如5而不是"5"），需要转成字符串再赋值：str(obj['priority'])
• putAttribute的第三个参数必须是字符串，这是NiFi API的强制要求

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