使用executestream命令的Python脚本

Question

在尽我最大的努力找到之前的问题和与这个问题相关的例子，但仍然没有找到我想要的答案之后，我想我会自己提交一个问题。

由于以下原因，ExecuteStreamCommand似乎是最适合我的处理器：

• 我能够执行任何Python脚本并避免Jython（以与execute script类似的方式）。Jython不是我的选择。
• 我可以接收流文件。这是必要的，因为我的脚本是使用前一个处理器的输出。此外，我喜欢将数据保存在“NiFi管理”下的想法。
• 它写一个“执行状态”，这将有助于路由。

简言之，我想用ExecuteStreamCommand做的是：

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

为了清楚起见，我现在不明白：

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

我遇到过许多ExecuteScript示例，但不幸的是，这些示例并不能完全转换为executestream命令的使用。

提前谢谢你。任何建议都很感激。

Answer 1

从您的问题中可以看出，您需要在不使用InvokeScriptedProcessor或ExecuteScript处理器的情况下调用Python脚本，因为您不能使用Jython。考虑到这一要求，你仍然应该能够实现你的目标。虽然它需要对框架有一定的了解，但所有这些信息都来自^{} documentation。

你的“我现在不明白”部分：

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

  • 在ExecuteStreamCommand处理器中，使用以下内容配置命令参数和命令路径属性：

    • 命令参数：any flags or args, delimited by ;（即/path/to/my_script.py）
    • 命令路径：/path/to/python3

• 如何从Python中加载流文件

  • 流文件内容将通过STDIN传递，因此在Python脚本中，处理数据的方式与通常处理STDIN的方式相同。
• 如何在Python中更新或创建新的流文件
  • NiFi在框架中处理流文件的创建。Python脚本传递给STDOUT的任何数据都将填充到传递给ExecuteStreamCommand处理器的输出流关系的结果流文件的内容中。在这种情况下，您的脚本不需要知道“流文件”。如果您使用的是ISP或ES处理器，则可以使用自动注入脚本的NiFi脚本API来创建或更新流文件对象。
• 如何将更新后的流文件从Python输出回NiFi。
  • 同样，只需从脚本中将所需的流文件内容写入到STDOUT，NiFi将使用该内容生成一个新的流文件。如果将ESC的输出目标属性属性设置为非空值，NiFi将使用包含脚本输出的同名新属性更新现有流文件。