在 Apache Nifi ExecuteStreamCommand 中运行 python 代码

Question

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

代码包含非纯 python 模块，如 Pandas 和 Numpy，因此不能使用 Nifi executeScript。

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

显然可以使用 STDIN 读取传入的流文件并使用 STDOUT 写出，请参阅这个 SO 问题： 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 读取然后修改内存中的数据，但从不将其写回。