使用ExecuteStreamCommand的Python脚本

Question

在尽力找到先前的问题和与该问题相关的示例之后，仍然找不到所需的答案，我想到自己会提交一个问题。

由于以下原因，ExecuteStreamCommand对我而言似乎是理想的处理器：

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

简而言之，我要对ExecuteStreamCommand进行的操作是：

• 摄取前一个处理器的输出（一个Scrapy Spider，它输出带有JSON行的文本文件）
• 调用python脚本（例如python3 my_script.py）
• 加载在我的python脚本中提取的FlowFile。
• 选择FlowFile的内容。
• 在python中操作FlowFile的内容。
• 输出原始FlowFile的更新版本或创建一个新版本。
• 使用更新的/新的FlowFile继续执行我的NiFi流。

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

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

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

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

Answer 1

根据您的问题，您说您需要在不使用InvokeScriptedProcessororExecuteScript处理器的情况下调用Python脚本，因为您无法使用Jython。鉴于此要求，您仍然应该能够实现目标。尽管需要一定程度地熟悉框架，但是所有这些信息都来自ExecuteStreamCommand文档。

您的“我目前不了解”部分：

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

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

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

  • 命令路径： /path/to/python3

  • 如何从Python内加载FlowFile

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

  • 如何在Python中更新或创建新的FlowFile
  • NiFi处理框架中的流文件创建。您的Python脚本传递给 STDOUT的 所有数据都将填充到传递给处理器的 输出流 关系的 结果流文件 的内容中ExecuteStreamCommand。在这种情况下，您的脚本不需要了解“流文件”。如果您改为使用ISP或ES处理器，则可以使用NiFi脚本API，该API将自动注入脚本中以创建或更新flowfile对象。
  • 如何将更新的FlowFile从Python输出回NiFi。
  • 同样，只需从脚本中将所需的流文件内容写入 STDOUT ，并且（返回状态代码为0）NiFi就会使用该内容生成一个新的流文件。如果将“ 输出目标属性” 属性设置ESC为非空值，NiFi将改为使用包含脚本输出的同名新属性来更新现有流文件。