如何在Nifi中使用MergeConent处理器

Question

我有一个Nifi流，在Json中有多行数据，对于每个行，需要执行以下操作：

• 拆分数据(EvaluateJsonPath)并将相关字段插入数据库(Mysql)，
• 将表A的字段取为表A，并将它们放入DB中。此操作的返回值是表A中自动生成的ID。
• 从表插入获取返回值，并与表B字段合并。表A的返回值成为表B中的引用值，因此分隔行条目成为链接。

我已经设置了flow (见图)，但是MergeContent处理器失败了，因为我认为它不知道如何正确地合并内容。如果我有200行，它只合并一行，其余的就失败了。我在Nifi中读过关于碎片整理的文章，但是没有成功地实现它，所以非常感谢您的任何输入。

流程图：

​

​

Answer 1

当您使用合并处理器时，流文件应该具有正确的属性，如fragment.index、fragment.identifier和fragment.count。如果您的流文件没有此功能，则合并将无法工作。

大多数情况下，这些属性是在拆分记录时自动创建的。因此，您只需在SplitAvro之前使用ConvertAvroToJson添加拆分过程。它将把查询结果一行行地分割成一行，并执行所需的操作，并将其合并。