使用 ExecuteStreamCommand 的 Python 脚本

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

由于以下原因，ExecuteStreamCommand 对我来说似乎是完美的处理器：

• 我能够执行任何 Python 脚本并避免使用 Jython（与 ExecuteScript 类似）。 Jython 不适合我。
• 我可以接受 FlowFiles。这是必要的，因为我的脚本是为了消耗先前处理器的输出而制作的。此外，我喜欢将数据保存在“NiFi 管理”之下的想法。
• 它写入对路由有用的“执行状态”。

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

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

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

• 如何调用 python 脚本（从 ExecuteStreamCommand 处理器）
• 如何从 Python 中加载 FlowFile
• 如何从 Python 中更新或创建新的 FlowFile
• 如何将更新后的 FlowFile 从 Python 输出回 NiFi。

我遇到过 ExecuteScript 的各种示例，但不幸的是，这些示例并不能完全转化为 ExecuteStreamCommand 的使用。

先感谢您。任何建议表示赞赏。

从你的问题来看，你说你需要调用Python脚本而不使用InvokeScriptedProcessor or ExecuteScript处理器，因为您无法使用 Jython。考虑到这一要求，您仍然应该能够实现您的目标。虽然需要对框架有一定的了解，但所有这些信息都来自ExecuteStreamCommand文档 https://nifi.apache.org/docs/nifi-docs/components/org.apache.nifi/nifi-standard-nar/1.5.0/org.apache.nifi.processors.standard.ExecuteStreamCommand/index.html.

您的“我目前不明白”部分：

• 如何调用 python 脚本（从 ExecuteStreamCommand 处理器）

  • In your ExecuteStreamCommand处理器，配置命令参数 and 命令路径具有以下属性：

    • 命令参数：any flags or args, delimited by ; (i.e. /path/to/my_script.py)
    • 命令路径：/path/to/python3

• 如何从 Python 中加载 FlowFile

  • 流文件内容将通过STDIN，因此在您的 Python 脚本中，以与通常处理方式相同的方式处理该数据STDIN.
• How to update or create a new FlowFile from within Python
  • NiFi 处理框架中的流文件创建。 Python 脚本传递给的任何数据STDOUT将填充到传递给的结果流文件的内容中输出流的关系ExecuteStreamCommand处理器。在这种情况下，您的脚本不需要了解“流文件”。如果您使用的是ISP or ES处理器，您可以使用 NiFi 脚本 API，它会自动注入到脚本中来创建或更新流文件对象。
• How to output the updated FlowFile from Python back to NiFi.
  • 同样，只需将所需的流文件内容写入STDOUT从您的脚本中，并且（给定返回状态代码0) NiFi 将生成包含该内容的新流文件。如果您设置输出目的地属性的财产ESC为非空值时，NiFi 将使用包含脚本输出的同名新属性来更新现有流文件。