如何在nifi中使用ExecuteScript和python从一个传入的流文件创建多个流文件
如何在nifi中使用ExecuteScript和python从一个传入的流文件创建多个流文件
提问于 2020-10-21 01:49:47
在本地运行，它的工作方式与我想要的完全一样(有一个传入的流文件，在位置7-10有许多不同的代码，每个唯一的代码输出一个文件)例如，如果记录1-5在位置7-10有1234，记录6在位置7-10有2345，记录7在位置7-10有1234，那么将有一个名为1234_file.txt的文件，其中有1-5和7行，第二个文件2345_file.txt将有来自输入文件的第6行：
f = open("test_comp.txt", "r")
for x in f:
comp = x[6:10]
print(comp)
n = open(comp+"_file.txt","a")
n.write(x)
n.close()
f.close()在nifi中，我尝试这样做：
from org.apache.commons.io import IOUtils
from java.nio.charset import StandardCharsets
from org.apache.nifi.processor.io import StreamCallback
class PyStreamCallback(StreamCallback):
def __init__(self):
pass
def process(self, inputStream, outputStream):
f = open(inputStream, 'r')
for x in f:
comp = x[6:10]
print("comp: ",comp)
newFile = open(comp+"_file.txt","a")
newFile.write(x)
flowFile = session.get()
if (flowFile != None):
flowFile = session.write(flowFile, PyStreamCallback())
session.transfer(flowFile, REL_SUCCESS)
session.commit()它似乎像预期的那样获得了输入，并正确地将comp存储在位置7-10，但我没有得到多个流文件(对于x6:10中的每个唯一字符串。出来的流文件是1个零字节文件。
你觉得我错过了什么？？
回答 1
Stack Overflow用户
发布于 2020-10-21 02:20:55
您将直接写入文件系统中的文件，而不是NiFi生态系统中的对象流文件。我建议阅读有关这些模式的上下文的Apache NiFi Developer's Guide，并查看一些Python ExecuteScript examples以查看相关的Python代码。
您需要创建多个flowfile对象，将数据映射到这些对象，然后将所有这些对象传输到相应的关系，而不是写出单个flowfile。
您是否需要使用自定义Python代码而不是SplitRecord和/或PartitionRecord处理器来完成此任务？我认为PartitionRecord会很容易地解决你所描述的问题。
页面原文内容由Stack Overflow提供。腾讯云小微IT领域专用引擎提供翻译支持
原文链接：
https://stackoverflow.com/questions/64450725
复制相关文章
相似问题
腾讯云开发者
