输入数据行作为python中的参数

Posted 2023-03-31

【中文标题】输入数据行作为python中的参数

【英文标题】：Input lines of data as arguments in python

【发布时间】：2019-09-24 22:35:21

【问题描述】：

我想使用NI-FI将数据从源传输到目标。在NI-FI我还想使用python脚本来屏蔽我的数据。所以有人建议我将数据的输入行一一传递作为参数蟒蛇。

所以我想知道我需要在我的 python 脚本中手动传递文件或每一行和每一列以进行屏蔽。什么是最好的方法，我怎样才能得到它。

在 NI-FI 中我也想使用 python 脚本来屏蔽我的数据，但是 Getfile Processor 和 python 脚本指出了位于 NI-FI 的 Input 文件夹中的同一个文件，所以可能是由于这个原因我在执行流命令中出现错误。

用户@GMc要求的以下脚本：-

   class ModJSON(StreamCallback):
    def __init__(self):
        pass

    def process(self, inputStream, outputStream):
        flowFile = session.get()
        obj = 
            "Source": "NiFi",
            "IP": obj['IP'].replace(regex='((?<=[0-9])[0-9]|(?<=\.)[0-9])', value='X'),
            "Name": obj['user']['screen_name']
        

在这个脚本中，我想屏蔽来自 GetFile 处理器但文件在队列中的数据。

我想从源中获取数据并对其进行转换。

【参考方案1】：

已经有一段时间了，但如果我没记错的话，NiFi 的 python 接口为您提供了一个名为 session 的全局变量。

您使用会话对象来获取流文件并将您的输出传输到正确的下游路径（例如成功）。

也许这个教程会对你有所帮助https://community.hortonworks.com/articles/35568/python-script-in-nifi.html

我对您对 GetInputFile（您的意思是 GetFile 处理器吗？）和 python 脚本的引用有点困惑。通常 GetFile 从文件系统中获取文件。然后将其转换为 NiFi 中称为 FlowFile 的内部机制。下游处理器（例如 python 脚本）从会话对象获取流文件对其进行处理，并可选择通过 session.transfer 方法传输修改后的版本。

【讨论】：

是的，它是 GetFile 处理器。我尝试使用你的方法，但我得到了另一个错误，即“无法获取宏（clojure）的值”你能帮我解决这个问题吗

您需要提供更多细节。例如，您是否尝试提供 clojure 脚本，或者您仍在尝试使用 python，但忘记在执行脚本处理器的“脚本引擎”下拉菜单中选择“pyton”。还有什么？？？

是的，我仍在使用 python，现在从“脚本引擎”中选择 python，但我的 nifi 程序从过去 1 小时开始仍在运行，并且文件在队列中。我想知道对于 python 脚本“executeStreamCommand”或“Eexecutescript”，哪个处理器更可取。因为我需要你的 Pandas 和 Json 在我的脚本中

再次，已经有一段时间了，但如果内存服务 ExecuteScript 为您提供了一个会话对象，允许您与 NiFi 集成（例如，选择传输路径，访问流文件等等）。而 ExecuteStreamCommand 没有。流文件是通过标准输入提供的。无论您选择哪种方式，只要您正确获取流文件，您就可以使用您可用的任何功能（例如 JSON 解析）处理内容。

现在我正在使用 ExecuteScript，在这个处理器中，我使用会话对象编写 python 脚本，但我的流程长时间运行，没有任何错误或任何输出，而且我的文件仍在队列中。你对此有什么想法吗？