9

新来的尼菲！

我想知道 nifi 中是否有一种方法可以使用诸如“PutFile”之类的处理器并将其写入一个文件（将数据附加到该文件，或覆盖该文件中的数据） - 而不是创建多个不同的文件？我需要使用另一个处理器来完成此操作吗？

4

5 回答 5

8

对于那些不想覆盖文件中的数据但想附加数据的人。

使用 ExecuteStreamCommand 处理器附加到单个文件：

putFile 处理器无法做到这一点，但您可以使用ExecuteStreamCommand处理器来完成此操作。

在命令参数中放置要记录的属性，由分隔符分隔 ${aatr1};${aatr2};${attr3}

在命令路径中放置 bash 脚本的绝对路径：/path/logger.sh

记录器.sh：

#!/bin/bash
echo "$1|$2|$3">> /path/attributes.log

attibutres.log 将逐行附加三个属性。确保 bash 脚本可以通过 nifi 执行。

做一个chmod 777 logger.sh

使用 ExecuteScript 处理器附加到单个文件：

试试这个 ECMAscript：

var flowFile = session.get();
var File = Java.type("java.io.RandomAccessFile");
if (flowFile != null) {
    var filename = flowFile.getAttribute("filename");
    /// write to file
    var filePath = "/path/attributes.log" ;
    flowFile = session.putAttribute(flowFile, "filePath", filePath);
     var file = new File(filePath, "rws");
     file.seek(file.length());
        file.write(filename.getBytes());
        file.write("\n".getBytes());
        file.close();
    // Finish by transferring the FlowFile to an output relationship
    session.transfer(flowFile, REL_SUCCESS);
}
于 2019-03-30T18:25:56.727 回答
3

目前无法将数据附加到文件，但您可以使用 PutFile 覆盖文件。

PutFile 处理器使用 FlowFile 上的属性“文件名”将文件写入磁盘。因此，如果您将 UpdateAttribute 处理器放在将所有传入 FlowFile 更新为相同“文件名”的 PutFile 之前，那么 PutFile 处理器将以相同的文件名将它们全部写入磁盘。

要使用 PutFile 执行此操作，请确保将处理器属性“冲突解决策略”配置为“替换”。

于 2016-05-31T00:03:05.577 回答
2

对不起。对此很陌生。让我再尝试一次...

LogAttribute 处理器可能是一个选项，因为它似乎提供了最接近文件附加的功能。但它似乎并不理想，因为它提供的指导输出的选项很少。

如果您打算使用“开箱即用”处理器功能而不是开发类来自定义，则可以尝试其他两个选项。使用 ExecuteScript 处理器执行 Jython、Groovy 或 JS 脚本，该脚本修改输出 flowFile 以仅包含您需要的属性。然后使用 PutMongo 或 PutSQL 处理器更新持久数据库资源。

如果您没有可供使用的数据库资源，则另一种选择：使用上面提到的 ExecuteScript 处理器，然后使用 PutFile 将唯一命名的文件名属性输出到目录 - 例如 ${filename}.${uuid}。您最终会得到大量类似格式的文件 - 每个文件一条日志记录 - 然后您可以将它们汇总到一个文件中以使用 Linux 行命令进行分析，或者在您的工作流程中使用最终的 ExecuteScript 处理器来每次汇总通过您的工作流程处理文件。最后一个可能不是一个好主意，因为如果您的 flowFiles 流很高，它是否会引入同步和写入争用问题尚不清楚。

于 2016-06-24T10:17:55.960 回答
0

有一种或多种将数据附加到文件的解决方案。我更喜欢 ExecuteGroovyScript 处理器来追加文件。您可以通过 groovy 脚本轻松获取文件。然后如果文件存在，我们可以附加数据。我成功地实现了这个方法。

于 2019-11-19T14:28:13.997 回答
0

最初的问题是附加到现有文件。大概的想法是为每个流文件写一行。

您可以使用不同的方法来实现此用例：不是将连续的行附加到文件中，而是在 Nifi 中进行合并，然后一次性写出整个文件。这是通过 MergeContent 处理器完成的。它获取连续流文件的内容，并将它们粘贴到单个流文件中。它有各种配置选项来说明在输出合并结果之前要累积多少个流文件，然后您可以使用 PutFile 将其写出。

这是一个以 CSV 格式将关键属性写入硬编码位置的方法。这对于在清空队列之前归档队列很有用。

假设您要写出属性id和。namesize

  • 添加处理器 AttributesToCSV。将“属性列表”设置为id,name,size。
  • 将“目标”设置为“流文件内容”。
  • 将其下游连接到 MergeContent。
  • 设置“最小/最大条目数”，以便您的整个队列适合一个箱（例如，分别为 1 和一些大数字）
  • 将“分隔符策略”设置为“文本”
  • 将“Demarcator”设置为换行符（类型⇧</kbd>+↩</kbd>, or paste in the newline)
  • 在“标头”中设置适当的 CSV 标头，例如“ID、名称、大小”并添加尾随换行符
  • 将“最大 bin 年龄”设置为几秒钟 - 足够长以便有时间合并所有内容，但足够短以让您乐于等待。
  • 将其连接到下游的 UpdateAttribute 将filename属性设置为您喜欢的任何内容
  • 将该下游连接到 PutFile，设置“目录”属性

其他一切都可以保留为默认值。

这个配方需要适应其他情况：

  • 如果有需要写出的恒定流：按条目数拆分为 bin，并使文件名动态化
  • 如果需要 CSV 以外的其他格式：将 AttributesToCSV 替换为其他处理器
于 2021-06-29T15:04:38.863 回答