Apache NiFi中CSV数据类型校验Python脚本第7行报错求助

原脚本存在的核心问题 #

• 作用域问题：session和flowFile无法在InputStreamCallback内部直接访问，必须通过构造函数传入
• 字符串验证逻辑无效：str(value)不会抛出ValueError，所有值都能转为字符串，该判断完全多余
• 无列数校验：未检查CSV行的列数是否与预期数据类型数组长度匹配，会引发索引越界错误
• FlowFile处理逻辑错误：在验证函数内直接操作flowFile和session会导致作用域混乱，且未正确处理会话提交
• 缺少会话提交：NiFi脚本必须调用session.commit()才能完成FlowFile的转移和属性设置

修改后的完整脚本 #

import csv
import io
from org.apache.commons.io import IOUtils
from org.apache.nifi.processor.io import InputStreamCallback
from java.nio.charset import StandardCharsets

class PyInputStreamCallback(InputStreamCallback):
    def __init__(self, session, flowFile):
        self.session = session
        self.flowFile = flowFile
    
    def process(self, inputStream):
        text = IOUtils.toString(inputStream, StandardCharsets.UTF_8)
        expected_data_types = ['int', 'str', 'str', 'str', 'str']  # 预期数据类型数组

        def is_valid_data_type(value, data_type):
            if data_type == 'int':
                try:
                    int(value.strip())  # 去除首尾空白后验证
                    return True
                except ValueError:
                    return False
            elif data_type == 'float':
                try:
                    float(value.strip())
                    return True
                except ValueError:
                    return False
            elif data_type == 'str':
                # 字符串类型无需验证，直接返回True
                return True
            else:
                return False
        
        def validate_csv(text, expected_data_types):
            reader = csv.reader(io.StringIO(text))
            next(reader)  # 跳过表头
            
            for row_idx, row in enumerate(reader, start=2):  # 行号从2开始（包含表头）
                # 检查列数是否匹配
                if len(row) != len(expected_data_types):
                    error_message = f"Invalid column count at row {row_idx}: expected {len(expected_data_types)}, got {len(row)}"
                    return error_message
                
                for col_idx, value in enumerate(row):
                    if not is_valid_data_type(value, expected_data_types[col_idx]):
                        error_message = f"Invalid data type at row {row_idx}, column {col_idx + 1}: '{value}' is not {expected_data_types[col_idx]}"
                        return error_message
            
            return None

        # 执行验证
        error_msg = validate_csv(text, expected_data_types)
        
        if error_msg:
            # 设置错误属性并转移到失败关系
            self.flowFile = self.session.putAttribute(self.flowFile, 'csv.validation.error', error_msg)
            self.session.transfer(self.flowFile, self.session.REL_FAILURE)
        else:
            # 验证通过，转移到成功关系
            self.session.transfer(self.flowFile, self.session.REL_SUCCESS)

# 获取FlowFile并处理
flowFile = session.get()
if flowFile is not None:
    session.read(flowFile, PyInputStreamCallback(session, flowFile))
    session.commit()
else:
    session.commit()

关键修改说明 #

1. 作用域传递：通过__init__方法将session和flowFile传入回调类，避免作用域访问错误
2. 简化字符串验证：直接返回True，因为任何值都可以转为字符串
3. 增加列数校验：在遍历行时先检查列数是否与预期一致，避免索引越界
4. 优化错误处理：让验证函数返回错误信息（而非直接操作session），在外部统一处理FlowFile的转移和属性设置
5. 添加会话提交：在处理完成后调用session.commit()，确保NiFi能正确处理FlowFile
6. 去除无用导入：删除了不需要的from org.apache.nifi.processors.script import ExecuteScript导入
7. 增强数值验证：验证前去除值的首尾空白，避免因空格导致的验证失败

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