NiFi - 将 hdfs 中的文件移动到文件目录属性

Question

我一直在尝试使用 MoveHDFS 处理器将 parquet 文件从 hdfs 中的 /working/partition/ 目录移动到 /success/partition/ 目录。分区值是基于流程中较早的 ExecuteSparkJob 处理器设置的。在根 / 目录中找到我的 parquet 文件后，我在 Output Directory 的处理器描述中找到了以下内容：

文件将被移动到的 HDFS 目录 Supports 表达式语言：true（将使用变量注册表进行评估 仅）

原来处理器将文件发送到/ 而不是${dir}/。

由于我的属性是根据 spark 处理结果动态设置的，因此我不能简单地添加到变量注册表并为每个流文件重新启动节点（根据我的有限理解，这是使用变量注册表所需要的）。一种选择是使用带有自定义脚本的 ExecuteStreamCommand 处理器来完成此用例。这是我唯一的选择，还是有一种内置方法可以将 HDFS 文件移动到属性集目录？

Answer 1

你可以试试这个方法：

第 1 步：使用 MoveHDFS 将文件移动到临时位置，例如路径 X。输入目录 MoveHDFS 处理器中的属性可以接受流文件属性。

第 2 步：成功连接到 FetchHDFS 处理器。

第 3 步：现在在 Fetch HDFS 处理器中，您可以将 HDFS 文件名 属性的表达式语言编写为 ${absolute.hdfs.path}/${filename}。这会将路径 X 中的文件数据提取到流文件内容中。

第 4 步：从 FetchHDFS 连接到 PutHDFS 处理器的成功连接。

第 5 步：根据您的要求配置 PutHDFS 目录属性以即时接受分区数据的流文件属性。

缺点： 这种方法的一个缺点是，将从 moveHDFS 创建的副本在将数据发送到实际位置之前临时存储数据。如果不需要，您可能必须开发一个单独的流程来删除重复的副本。