nifi ExecuteScript使用
程序员文章站
2022-07-06 16:20:02
...
本文使用python,其他语言请参考这里
1.从FlowFile获取文件:
flowFile = session.get()
2.脚本结束后路由(不路由会报错)
#路由到成功
session.transfer(flowFile, REL_SUCCESS)
#路由到失败
session.transfer(flowFile, REL_FAILURE)
3.设置文件属性
flowFile = session.putAttribute(flowFile, 'attr', 'attr')
4.完整代码
from org.apache.commons.io import IOUtils
from java.nio.charset import StandardCharsets
from org.apache.nifi.processor.io import StreamCallback
flowFile = session.get()
class PyStreamCallback(StreamCallback):
def __init__(self):
pass
def process(self, inputStream, outputStream):
text = IOUtils.toString(inputStream, StandardCharsets.UTF_8)
#正文
outputStream.write(bytearray(text.encode('utf-8')))
if(flowFile != None):
flowFile = session.write(flowFile, PyStreamCallback())
flowFile = session.putAttribute(flowFile, 'attr', 'atr')
session.transfer(flowFile, REL_SUCCESS)