我一直在尝试从 nifi 的 ExecuteScript 处理器中提取数据并将其作为属性附加到流文件。我尝试了很多来源,尤其是 Matt Burgess 的 funnifi 博客上的一个。
以下是我的代码
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
originalFlowFile = session.get()
text = IOUtils.toString(originalFlowFile)
log.info(text)
if(originalFlowFile != None):
event = json.loads(text)
if (event['true'] == 'Y'):
flowfile = session.putAttribute(flowfile, "true", "Y")
elif (event['src'] == 'ONE' ):
allAttributes = { "true": "N", "src": "ONE" }
flowfile = session.putAllAttributes(flowfile, allAttributes)
elif (event['src'] == 'TWO' ):
allAttributes = { "true": "N", "src": "TWO" }
flowfile = session.putAllAttributes(flowfile, allAttributes)
session.transfer(flowFile, REL_SUCCESS)
session.commit()
这是用Python编写的。流文件数据是 JSON。尽管如此,我还是无法解析它。
INFO部分是输出
text = IOUtils.toString(originalFlowFile)
任何帮助将不胜感激。
附:我对 python 不熟悉
测试数据
{
"true":"N",
"src":"ONE",
"var1":"value1",
"var2":"value2"
}
更新
我更新的代码仍然不起作用:
import json
import java.io
from org.apache.commons.io import IOUtils
originalFlowFile = session.get()
if(originalFlowFile != None):
inputStream = session.read(originalFlowFile)
text = IOUtils.toString(inputStream)
log.info(text)
event = json.loads(text)
if (event['true'] == 'Y'):
flowfile = session.putAttribute(flowfile, "true", "Y")
elif (event['src'] == 'ONE' ):
allAttributes = { "true": "N", "src": "ONE" }
flowfile = session.putAllAttributes(flowfile, allAttributes)
elif (event['src'] == 'TWO' ):
allAttributes = { "true": "N", "src": "TWO" }
flowfile = session.putAllAttributes(flowfile, allAttributes)
session.transfer(flowFile, REL_SUCCESS)
session.commit()
最佳答案
对流文件本身调用 IOUtils.toString() 可能不起作用,因为流文件不是 InputStream、Reader 或可以自行读取的文件。事实上,我很惊讶该行没有产生异常。
有两种方法可以获取流文件的内容...
第一个方法是从 session 中获取流文件的 InputStream:
originalFlowFile = session.get();
inputStream = session.read(originalFlowFile);
text = IOUtils.toString(inputStream);
第二种是使用InputStreamCallback:
flowFile = session.read(flowFile, {inputStream ->
// read the inputStream
} as InputStreamCallback);
关于java - Nifi 在 ExecuteScript 中遇到错误,我们在Stack Overflow上找到一个类似的问题: https://stackoverflow.com/questions/46115607/