Nifi Python脚本报list indices must be integers错误，求排查建议

错误点梳理 #

• 核心触发报错的问题：你传入的JSON内容是数组结构，json.loads(text)执行后得到的data是列表类型，代码中newObj = data['priority']用字符串作为列表的索引，不符合列表的取值规则，直接触发list indices must be integers报错。你本地测试没有报错大概率是本地测试时要么注释了这行代码，要么用的测试JSON是对象结构而非数组。
• 输出流写入逻辑错误：计算完成后的finalResp是列表类型，没有encode方法，需要先序列化为JSON字符串才能写入输出流。
• 属性读取逻辑错误：如果你要获取的priority是FlowFile的属性，需要从外层的flowFile对象读取，不能从FlowFile的内容数据中提取。
• filename属性赋值错误：原代码中将自定义类的实例转为字符串赋值给filename，得到的是无意义的对象标识，不符合业务要求。

修改后代码 #

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, priority):
        # 把从外层拿到的priority属性传入类内部使用
        self.priority = priority
        pass
    def process(self, inputStream, outputStream):
        finalResp = []
        text = IOUtils.toString(inputStream, StandardCharsets.UTF_8)
        data = json.loads(text)
        # 移除错误的data['priority']取值逻辑
        newObj = self.priority
        for k in data:
            resp = self.performCalculation(k)
            finalResp.append(resp)

        log.info(" newObj ",newObj)
        # 先将结果列表转为JSON字符串，再编码写入
        output_str = json.dumps(finalResp)
        outputStream.write(bytearray(output_str.encode('utf-8')))

    def performCalculation(self,k):
            a = int(k['a'])
            b = int(k['b'])
            log.info(a)
            log.info(b)
            total=sum((a,b))
            log.info(total)
            x = {"x":total}
            k.update(x)
            return k

flowFile = session.get()
if flowFile != None:
    # 可按需打开设置priority属性的逻辑
    flowFile = session.putAttribute(flowFile, "priority", "5")
    # 从flowFile属性中读取priority
    priority_val = flowFile.getAttribute('priority')
    # 将priority传入自定义处理类
    parser = FlowFileParser(priority_val)
    session.write(flowFile, parser)
    # 按需修改filename的赋值规则，这里示例保留原文件名加后缀
    origin_filename = flowFile.getAttribute('filename')
    flowFile = session.putAttribute(flowFile, "filename", f"calculated_{origin_filename}")
    session.transfer(flowFile, REL_SUCCESS)

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