如何在 NiFi 的 ExecuteStreamCommand 处理器中读取文件答案

【问题标题】：How to read files in ExecuteStreamCommand processor in NiFi如何在 NiFi 的 ExecuteStreamCommand 处理器中读取文件
【发布时间】：2021-01-21 15:51:03
【问题描述】：

我的最终目标是屏蔽一个特定文件中的数据。我想将文件从一个地方移动到另一个地方。在此传输过程中，我必须使用 Python 脚本屏蔽数据。所以，我设计了以下流程：

GetFile > ExecuteStreamCommmand > PutFile

我使用pandas 设计了一个Python 脚本。我在Google Cloud Platform 创建的虚拟机上运行这个NiFi，我安装了Python-2.7 和NiFi-1.9.1。以下是我的熊猫代码：

import pandas as pd
readFile = pd.read_csv("/path",sep=" ",header=None)
readFile.columns = ['IP']
readFile['IP'] = readFile['IP'].replace(regex='((?<=[0-9])[0-9]|(?<=\.)[0-9])',value='X')
readFile.to_csv("/path", sep=' ')

我有以下疑问：
1) 使用 getFile 处理器，我将队列中的文件传递给下一个处理器，即 ExecuteStreamCommand 处理器。
2）另外，在我的 Python 代码中，我试图从在 GetFile 处理器中传递的同一输入目录中读取数据，但现在该文件已移动到 getfile > executestreamcommand 之间的队列中。那么它将如何读取呢？
3）python脚本执行后如何使用putFile处理器将其保存回其他地方？

我是 NiFi 的新手，所以试图了解基本的东西。另外，我附上了流程和错误截图。

【问题讨论】：

你在“Python”代码中做了什么？看来您只是替换了文件中的某些内容，对吗？我们无法在执行流命令处理器中读取传入的流文件
但我想在 executestreamcommand 处理器中使用正则表达式屏蔽数据。那我该怎么做呢？这种情况下的流程是什么？
看来您可以在 ReplaceText 处理器本身中屏蔽数据，无需使用 Python 脚本。

标签： python hadoop google-cloud-platform apache-nifi

【解决方案1】：

ExecuteStreamCommand

流文件的内容作为stdin流传递给命令（在您的情况下为python）

所以，你必须使用以下代码：

readFile = pd.read_json(sys.stdin)

另一方面，如果您需要对流文件应用正则表达式替换，您可以尝试使用ReplaceText 处理器而不是ExecuteStreamCommand

【讨论】：

谢谢@daggett。我有柱状数据。那么替换文本处理器对我的目的来说是一个不错的选择吗？另外，我可以使用 read_csv 方法读取数据，因为我想提到分隔符吗？
pandas 可以在 jython 上工作吗？
@YangwookIanJeong 不，pandas 不能在 jython 中工作，因为它是一个本地编译的 python 库

【解决方案2】：

您可能需要在已注册 nifi 的 Volume 中提供 python .py 文件。

例如 opt/nifi/nifi-current/ 如果是 docker 镜像

【讨论】：