在 apache-nifi 中使用 executescript 处理器更新 csv 值失败

Question

我尝试从流文件中读取并使用 csv 中的默认值更新记录值。为此，我使用了 ExecuteScript 处理器，其中包含以下 python 代码。

import sys
import re
import traceback
from org.apache.commons.io import IOUtils
from org.apache.nifi.processor.io import StreamCallback
from org.python.core.util import StringUtil
from java.lang import Class
from java.io import BufferedReader
from java.io import InputStreamReader
from java.io import OutputStreamWriter

flowfile = session.get()
record = flowfile.getAttribute('record_type')

if record == '0':
    flowfile = session.putAttribute(flowfile,'record_type', 'NEW_USER')
    session.transfer(flowFile, REL_SUCCESS)
    session.commit()
elif record == '1':
    flowfile = session.putAttribute(flowfile,'record_type', 'OLD_USER')
    session.transfer(flowFile, REL_SUCCESS)
    session.commit()
else:
    flowfile = session.putAttribute(flowfile,'record_type', 'IGNORE')
    session.transfer(flowFile, REL_SUCCESS)
    session.commit()

writer.flush()
writer.close()
reader.close()

我的 csv 看起来像

id,record_type
1,0
2,1
3,2
4,0

结果应该是：

id,record_type
1,NEW_USER
2,OLD_USER
3,IGNORE
4,NEW_USER

我收到以下错误：

AttributeError : 'NoneType' 对象在中没有属性 'getAttribute' 第 13 行的脚本

上面写着record = flowfile.getAttribute('record_type')这是错误的..

我不知道如何解决这个问题，因为我不擅长 python。

Answer 1

这不是 python，但根据作者的评论可能是 groovy。

使用带有以下代码的 ExecuteGroovyScript 处理器：

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

def map = [
    '0': 'NEW_USER',
    '1': 'OLD_USER',
]

ff.write{rawIn, rawOut->
    rawOut.withWriter("UTF-8"){w->
        rawIn.withReader("UTF-8"){r->
            int rowNum = 0
            //iterate lines from input stream and split each with coma
            r.splitEachLine( ',' ){row->
                if(rowNum>0){
                    //if not a header line then substitute value using map
                    row[1] = map[ row[1] ] ?: 'IGNORE'
                }
                //join and write row to output writer
                w << row.join(',') << '\n'
                rowNum++
            }
        }
    }
}

REL_SUCCESS << ff