python - 在 Apache Nifi ExecuteStreamCommand 中运行 python 代码

Question

我正在尝试在 Nifi ExecuteStreamCommand 处理器中运行 python 代码。

代码包含非纯Python模块，例如Pandas和Numpy，因此不能选择使用NifiexecuteScript。

问题在于读取流文件和修改流文件内容。

显然可以使用 STDIN 读取传入流文件并使用 STDOUT 写出，请参阅此问题: Python Script using ExecuteStreamCommand

但我无法让这个工作。

1.尝试简单地从 STDIN 读取 CSV 并对其进行修改，但当发送到 putFile 处理器时，文件是相同的。

import sys
import pandas as pd
import io

df = pd.read_csv(io.StringIO(sys.stdin.read(1)))
df2 = pd.DataFrame([[5, 6], [7, 8]], columns=list('AB'))
df2 = df.append(df2)

2.尝试将一些其他代码包装在函数中并假设函数输出将转到 STDOUT，但返回相同的结果。

def convert_csv_dataframe():
    a = pd.read_csv(io.StringIO(sys.stdin.read(1)))
    a.replace(["ABC", "AB"], "A", inplace=True)
    return a

convert_csv_dataframe()

如果有人可以提供帮助，我们将不胜感激。

编辑:

这段代码有效。问题出在 Nifi 上。我正在阅读“原始”关系而不是“输出流”关系。请注意，标准输入正在读取一行，但认为这不会产生影响。我唯一的问题是:我可以从executeStreamCommand引用流文件本身(而不是它的内容)吗？

import sys

a = sys.stdin.readline()
a = a.upper()
sys.stdout.write(a)

Answer 1

我认为您需要在脚本中的某个位置写入 STDOUT。我不太了解 Python，但这两个示例看起来都像是从 STDIN 读取，然后修改内存中的数据，但永远不会将其写回。