MergeContent

简介

MergeContent Processor 提供了把多个 FlowFile 合并为单个 FlowFile 的能力。数据流设计者可能有很多 理由这样做。例如，在发送到下游系统之前把数据组成批次可能很有帮助，因为下游系统对大文件的处理 往往比大量小文件更优化。Lynxus 本身也能从中受益，因为 Lynxus 在"微批处理（micro-batches）" 模式下运行效果最佳，即每个 FlowFile 的大小在数 KB 到数 MB 之间。

该 Processor 会创建若干个"箱（bin）"来放置 FlowFile。默认最多使用 5 个箱，但可以通过更新 属性的值来修改。限制箱的数量是为了避免耗尽 Java 堆空间。注意： 虽然 FlowFile 的内容存储在 Content Repository 而不是 Java 堆空间中，但该 Processor 必须把 FlowFile 对象本身保存在内存里。因此，这些带着 Attribute 的 FlowFile 可能占用大量堆空间， 导致抛出 OutOfMemoryError。为避免这种情况，如果预计要合并大量小 FlowFile，建议先用一个 MergeContent 把不超过（比如）1,000 个 FlowFile 合并成小包，再用第二个 MergeContent 把这些 小包合并成更大的包。例如，要合并 1,000,000 个 FlowFile，可以使用 为 1,000 的 MergeContent，并把 "merged" Relationship 路由到第二个同样设置 为 1,000 的 MergeContent。第二个 MergeContent 会把 1,000 个 含 1,000 条数据的小包合并，实际上就产生了含 1,000,000 条数据的包。

FlowFile 如何装箱

Processor 决定把 FlowFile 放入哪个箱，取决于若干配置选项。首先是 Merge Strategy。 Merge Strategy 可以设置为两个选项之一："Bin Packing Algorithm"（装箱算法）或 "Defragment" （碎片重组）。如果目标只是把较小的 FlowFile 合并成一个更大的 FlowFile，应使用 Bin Packing Algorithm。该算法根据以下条件选择箱：按大小和 属性判断 FlowFile 是否能放进该箱，以及该 FlowFile 与箱中其他 FlowFile 是否"相似"。两个 "相似（like）FlowFile"的含义将在本节末尾讨论。

当需要把 FlowFile 显式地分配到同一个箱时，可以使用 "Defragment" 合并策略。例如，如果用 UnpackContent Processor 拆分了数据，每个拆出的 FlowFile 可以独立处理，之后再用本 Processor （Merge Strategy 设为 Defragment）合并回来。要在这种配置下把 FlowFile 加入同一个箱， 这些 FlowFile 的 "fragment.identifier" Attribute 必须具有相同的值。具有相同标识符的每个 FlowFile 还必须有唯一的 "fragment.index" Attribute 值，以便正确排序。对于给定的 "fragment.identifier"，至少要有一个 FlowFile 带有 "fragment.count" Attribute（表示箱中 应包含多少个 FlowFile）。具有相同标识符的其他 FlowFile 要么与该 "fragment.count" Attribute 的值相同，要么可以省略该属性。注意： 虽然拆分后再重新合并有合理的使用场景， 但把一个较大的 FlowFile 拆成上百万个小 FlowFile 再重新合并是一种反模式。这样做可能消耗大量 Java 堆并导致 Out Of Memory 错误，还会给 Lynxus 框架带来很大负载，造成 CPU 和磁盘占用升高， 吞吐量往往会下降一个数量级、延迟升高一个数量级。作为替代方案，数据流应尽可能使用面向 Record 的 Processor，例如 QueryRecord、PartitionRecord、UpdateRecord、LookupRecord、 PublishKafkaRecord_2_6 等。

要被放入同一个箱，两个 FlowFile 必须是"相似（like）的"。两个相似 FlowFile 必须具有相同的 schema；如果设置了 属性，它们的指定 Attribute 还必须具有 相同的值。例如，如果 设为 "filename"，那么两个 FlowFile 的 "filename" Attribute 必须相同才能装进同一个箱。如果需要用多个 Attribute 来关联两个 FlowFile，建议在 MergeContent 之前使用 UpdateAttribute Processor 把这些 Attribute 组合起来。例如，如果目标只在 "abc" 和 "xyz" 两个 Attribute 的值都相同时才把两个 FlowFile 装进同一个箱，那么可以用 UpdateAttribute 添加一个名为 "correlation.attribute"、 值为 "abc=${abc},xyz=${xyz}" 的属性，然后把 MergeContent 的 属性设为 "correlation.attribute"。

箱何时合并

上面讨论了如何为给定的 FlowFile 选择箱。一旦创建了箱并向其中添加了 FlowFile，就需要某种方式 判断箱何时"满了"，从而把这些 FlowFile 合并成一个 "merged" FlowFile。

如果 属性设为 "Bin Packing Algorithm"，则会评估以下规则。

MergeContent 暴露了若干不同的阈值，可用于创建大小理想的箱。例如，用户可以指定合并前必须打包在 一起的最少 FlowFile 数量，也可以配置最小字节数；此外，还可以指定最大 FlowFile 数量和字节数。

还有两种其他条件会导致箱内容被合并。Max Bin Age 属性指定了箱被合并之前，FlowFile 可以一起 装箱的最长时间。这个属性几乎总是应该设置，因为它相当于为箱设置了超时，这样即使数据暂时停止流入 该 Processor（由于上游系统问题、源 Processor 被停止等），FlowFile 也不会无限期地滞留在 MergeContent Processor 中。此外，该 Processor 还暴露了一个属性用于指定应使用的最大箱数量。 对某些用例来说这并不重要；但如果设置了 Correlation Attribute 属性，这就很重要了。当传入的 FlowFile 需要被放入某个箱时，Processor 必须找到合适的箱，否则就要创建新箱。如果必须创建新箱， 而现有的箱数量已经达到 属性的值，那么最旧的箱会被合并以腾出空间。

如果 属性设为 "Defragment"，那么只有当箱中 FlowFile 的数量等于箱内某个 FlowFile 的 "fragment.count" Attribute 指定的数量时，箱才算满。所有带有该 Attribute 的 FlowFile 必须具有相同的值，否则会被路由到 "failure" Relationship。并不要求所有 FlowFile 都有这个值，但箱内至少要有一个，否则箱永远无法装满。如果在箱超时（由 属性 指定）之前所有必要的 FlowFile 没有装到同一个箱，这些 FlowFile 会被全部路由到 'failure' Relationship，而不是合并。

最后，如果配置了 属性，且收到的某个 FlowFile 满足指定条件，箱也可以 被合并。该条件以表达式语言（Expression Language）表达式指定。如果任何 FlowFile 使该表达式 返回 true，无论箱中有多少数据、箱已存在多久，箱都会被合并。这个触发合并的传入 FlowFile， 根据 属性的值，可以作为箱的最后一个条目加入、作为新箱的第一个 条目，或者作为独立的箱输出。

因此，当满足以下任意一个条件时，一箱 FlowFile 就会被合并：

• 箱已达到最大字节数，由 属性配置。
• 箱已达到最大 FlowFile 数，由 属性配置。
• 箱已同时达到最小字节数（由 属性配置）和最小 FlowFile 数（由 属性配置）。
• 箱已达到最大存活时间，由 属性配置。
• 已达到最大箱数量（由 属性配置），且必须创建新箱。
• 配置了 属性，且收到的某个 FlowFile 满足指定条件。

合并原因（Reason for Merge）

每当箱的内容被合并时，合并产生的 FlowFile 上都会添加一个名为 "merge.reason" 的 Attribute。 下表列出了该属性的所有可能取值及各自的含义。

注意，Attribute 的值命名得很简洁，而文字说明则详尽得多。这样做有几个原因：首先，存储较大的 Attribute 值成本更高，会占用更多堆空间并需要更多资源处理；其次，简洁的取值更便于表述；最重要的是， 这意味着必要时可以使用 RouteOnAttribute 这类 Processor 基于该 Attribute 的值进行路由。这样一来， 说明文字可以进一步扩充或更新，而无需改变 Attribute 的取值。