Nifi：使用 MergeContent 处理器一次性合并所有必要的流文件答案

【问题标题】：Nifi: Merging all of necessary flowfile in one shot with MergeContent ProcessorNifi：使用 MergeContent 处理器一次性合并所有必要的流文件
【发布时间】：2019-05-29 08:26:59
【问题描述】：

这是关于 Nifi 中 MergeContent 处理器的问题。目前，我需要以一种方式将所有流文件与一个特定属性结合起来。

但是会发生什么，因为有这么多具有相同属性的流文件，处理器产生几个与属性合并的不同流文件，这几个应该组合成一个大的流文件。

他们有什么方法可以等待所有具有相同属性的流文件进入 MergeContent 处理器并一次合并它们？

如果没有，还有什么替代方法？

【问题讨论】：

您有等待的文件数吗？或者 nifi 应该等待具有相同属性的文件的任何其他限制？
不，我没有文件数。它不时根据输入文件的大小而变化。输入文件被一些标签（使用 EvaluateXQuery）分割成大量不同的流文件。所以有多少个属性值相同的小文件是未知的。或者我能以某种方式知道吗？

标签： merge apache-nifi

【解决方案1】：

您可以增加“最小条目数”和“最大条目数”以适合您的使用情况，并将“最大 Bin Age”增加到合适的时间。

您可以在此处参考合并内容的所有属性：https://nifi.apache.org/docs/nifi-docs/components/org.apache.nifi/nifi-standard-nar/1.6.0/org.apache.nifi.processors.standard.MergeContent/index.html

另外，您是在集群中使用 nifi 还是单独使用？因为一个节点中的流文件无法与另一节点中的流文件合并。

例如，如果您使用具有 5 个节点的 nifi 集群，每个节点有 10 个流文件。使用合并内容处理器后，您将拥有 5 个文件。

如果您需要进一步的帮助，请告诉我！

【讨论】：

我单独使用它。但问题是一个大的输入 xml 文件（也嵌套很深）到一个节点中将被拆分为 1000 到 1500 个不同的流文件（在此处展平），以便每个流文件都可以转换为 csv。我在这里使用 MergeContent 来组合所有共享相同特定属性的小型 csv 流文件。所以，不幸的是，我无法设置最小条目数，因为条目数绝对因文件而异。
那么在这种情况下，我认为增加最大 bin 年龄可能会有所帮助。您可以将其增加到 120 秒。除此之外，您可以将最小条目数设置为更高的数字，这样您就可以确保首先满足哪个条件，都会被执行
合并后的流文件去哪了？如果它的 hdfs，我认为有一种解决方法，您可以使用“更新属性”处理器重命名所有流文件，然后在 PutHDFS 处理器中，您可以使用冲突解决策略作为“附加”。这意味着所有具有相同名称的流文件都将附加到 HDFS 中的同一个文件中。（PS：我不太确定这一点，建议您测试一次，确保将页脚设置为空行以避免数据附加到同一行）。
嗯，要求是 Nifi 管道需要将 xml 转换为 csv 理想情况下每个文件 1 秒，即使不确定是否可能。所以我也不能将最大 bin 年龄设置为 120 秒。对于 HDFS，由于新的 csv 表具有不同的列数，因此简单地追加可能不起作用。（可能需要实现自定义脚本将新表合并到现有表）

【解决方案2】：

在 MergeContent 处理器之前，您应该使用“单节点”的负载平衡策略配置您的连接器，这样您就可以保证只有 1 个节点会处理碎片整理。

【讨论】：