使用 ExecuteStreamCommand 的 Python 脚本 [英] Python Script using ExecuteStreamCommand

问题描述

在尽我所能找到与此问题相关的先前问题和示例后，仍然没有找到我正在寻找的答案，我想我会自己提交一个问题.

After doing my best to find previous questions and examples relevant to this question, and still not finding the answers that I'm looking for I figured that I would submit a question myself.

ExecuteStreamCommand 对我来说似乎是完美的处理器，原因如下:

ExecuteStreamCommand seems like the perfect processor for me due to the following reasons:

• 我能够执行任何 Python 脚本并避免使用 Jython(以与 ExecuteScript 类似的方式).Jython 不适合我.
• 我可以接受 FlowFiles.这是必要的，因为我的脚本是用来消耗前一个处理器的输出的.此外，我喜欢将数据保存在NiFi 管理"之下的想法.
• 它会写入一个执行状态"，这对路由很有用.

简而言之，我试图用 ExecuteStreamCommand 做的是:

In a nutshell, what I'm trying to do with ExecuteStreamCommand is:

• 摄取前一个处理器的输出(一个 Scrapy 蜘蛛，准确地输出带有 JSON 行的文本文件)
• 调用 python 脚本(例如 python3 my_script.py)
• 加载在我的 Python 脚本中提取的 FlowFile.
• 选择 FlowFile 的内容.
• 在 python 中操作 FlowFile 的内容.
• 输出原始 FlowFile 的更新版本或创建一个新版本.
• 使用更新的/新的 FlowFile 继续我的 NiFi 流程.

为了清楚起见，我目前不明白:

For clarity's sake I currently don't understand:

• 如何调用 python 脚本(来自 ExecuteStreamCommand 处理器)
• 如何从 Python 中加载 FlowFile
• 如何在 Python 中更新或创建新的 FlowFile
• 如何将更新后的 FlowFile 从 Python 输出回 NiFi.

我遇到了 ExecuteScript 的各种示例，但不幸的是，这些并没有完全转化为 ExecuteStreamCommand 的使用.

I have come across various examples for ExecuteScript, but unfortunately these don't exactly translate to the use of the ExecuteStreamCommand.

先谢谢你.任何建议表示赞赏.

Thank you in advance. Any advice is appreciated.

推荐答案

根据您的问题，您说您需要在不使用 InvokeScriptedProcessor 或 ExecuteScript 处理器的情况下调用 Python 脚本因为你不能使用 Jython.鉴于该要求，您仍然应该能够实现您的目标.虽然它需要对框架有一定的了解，但所有这些信息都来自 ExecuteStreamCommand 文档.

From your question you say you need to invoke the Python script without using the InvokeScriptedProcessor or ExecuteScript processors because you can't use Jython. Given that requirement, you should still be able to accomplish your goal. While it requires some familiarity with the framework, all of this information is from the ExecuteStreamCommand documentation.

您的我目前不明白"部分:

Your "I currently don't understand" section:

• 如何调用 python 脚本(来自 ExecuteStreamCommand 处理器)

• How to call the python script (from the ExecuteStreamCommand Processor)

• 在您的 ExecuteStreamCommand 处理器中，使用以下内容配置 Command Arguments 和 Command Path 属性:

• In your ExecuteStreamCommand processor, configure the Command Arguments and Command Path properties with the following:

• 命令参数:任何标志或参数，以;分隔(即/path/to/my_script.py)
• 命令路径:/path/to/python3

如何从 Python 中加载 FlowFile

How to load up the FlowFile from within Python

• 流文件内容将通过 STDIN 传递，因此在您的 Python 脚本中，以与通常处理 STDIN 相同的方式处理该数据.

• The flowfile content will be passed via STDIN, so in your Python script, process that data the same way you would normally process STDIN.

• NiFi 处理框架中的流文件创建.您的 Python 脚本传递给 STDOUT 的任何数据都将填充到传递给 ExecuteStreamCommand 处理器的输出流关系的结果流文件的内容中.在这种情况下，您的脚本不需要对流文件"有任何了解.如果您使用 ISP 或 ES 处理器，则可以使用自动注入脚本的 NiFi 脚本 API 来创建或更新流文件对象.

• NiFi handles the flowfile creation in the framework. Any data passed by your Python script to STDOUT will be populated into the content of the resulting flowfile passed to the output stream relationship of the ExecuteStreamCommand processor. Your script does not need to have any awareness of "flowfiles" in this instance. If you were instead using the ISP or ES processors, you could use the NiFi scripting API which is automatically injected into the scripts to create or update the flowfile object.

• 同样，只需从脚本中将所需的流文件内容写入 STDOUT，并且(给定返回状态代码 0)NiFi 将生成一个包含该内容的新流文件.如果您将 ESC 的 Output Destination Attribute 属性设置为非空值，NiFi 将改为使用包含输出的同名新属性更新现有流文件剧本.

• Again, simply write the desired flowfile content to STDOUT from your script, and (given a return status code of 0) NiFi will generate a new flowfile with that content. If you set the Output Destination Attribute property of ESC to a non-null value, NiFi will instead update the existing flowfile with a new attribute of the same name containing the output of the script.

这篇关于使用 ExecuteStreamCommand 的 Python 脚本的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！