MergeRecord

简介

MergeRecord Processor 允许用户把多个由面向记录的数据（任何存在可用 Record Reader 的数据格式） 组成的 FlowFile 合并为一个更大的 FlowFile。在把数据推送到偏好大批次数据的下游系统（例如 HDFS）之前，这样做可能更好；或者为了通过减少系统中流转的 FlowFile 数量来提升 Lynxus 流程的 性能（从而降低 FlowFile Repository、Provenance Repository、Content Repository 和 FlowFile 队列上的竞争压力）。

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

当需要把记录显式地分配到同一个箱时，可以使用 "Defragment" 合并策略。例如，如果用 SplitRecord Processor 拆分了数据，每个拆分结果可以独立处理，之后再用本 Processor （Merge Strategy 设为 Defragment）合并回来。要在这种配置下把 FlowFile 加入同一个箱， 这些 FlowFile 的 "fragment.identifier" Attribute 必须具有相同的值。具有相同标识符的每个 FlowFile 还必须具有相同的 "fragment.count" Attribute 值（表示箱中应包含多少个 FlowFile）以及唯一的 "fragment.index" Attribute 值，以便正确排序。

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

只把某个字段值相同的记录装进同一个箱往往很有用。例如，如果有销售点（POS）数据，也许希望把 属于同一门店（由 'storeId' 字段标识）的记录装在一起。这可以通过在 MergeRecord 之前使用 PartitionRecord Processor 来实现。该 Processor 允许把一个或多个字段配置为分区标准， 并会为对应的值创建 Attribute。如有需要，还可以用 UpdateAttribute Processor 把多个 Attribute 组合为单个关联属性（如上所述）。更多细节请参阅这些 Processor 的文档。

箱何时合并

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

如果 属性设为 "Bin Packing Algorithm"，则会评估以下规则。首先，箱要算满， 必须同时满足 和 属性指定的两个阈值； 如果其中一个属性未设置，则忽略它。其次，如果达到 或 属性中的任意一个，箱就会被合并。也就是说，两个最小值都必须 达到，但最大值只需达到其中一个。注意， 属性是"软限制"，意味着 给定输入 FlowFile 中的所有记录都会被加入同一个箱，因此记录数量可能超过配置的最大记录数。 但一旦发生这种情况，就不会再从其他 FlowFile 向该箱添加记录。如果某个箱达到 ， 该箱中的 FlowFile 会被合并，即使尚未满足最小箱大小和最小记录数。最后，如果已创建的箱 数量达到最大值（由 属性指定），且某些输入 FlowFile 放不进任何 现有的箱，那么最旧的箱会被合并以腾出空间。这样做是因为否则将无法向现有箱添加任何新的 FlowFile，只能等到达到 Max Bin Age（如果真能等到的话）才能合并。

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

箱合并成单个 FlowFile 之后，有时了解箱为何在那个时刻被合并会很有用。例如，如果达到允许的 最大箱数量，合并产生的 FlowFile 可能只包含远少于预期数量的记录。为帮助理解这种行为， 该 Processor 在创建合并的 FlowFile 时会发出 JOIN Provenance 事件，该事件包含一个 "Details" 字段，解释箱为何在那个时刻被合并。例如，如果箱达到了最小阈值且之后没有再添加 FlowFile，事件会显示 "Records Merged due to: Bin is full"；如果箱是因为配置的最大箱数量 已满、需要为新箱腾出空间而被合并，事件则可能显示 "Records Merged due to: Maximum number of bins has been exceeded"。

失败发生时

当一个箱被装满时，Processor 负责把这些 FlowFile 中的所有记录合并成单个 FlowFile。如果 Processor 因任何原因失败（例如无法从某个输入 FlowFile 中读取记录），该箱中的所有 FlowFile 都会被路由到 'failure' Relationship。Processor 不会跳过出问题的那个 FlowFile 而合并其他的。选择这种行为是出于两点考虑：首先，缺少了那些问题记录，箱可能并不 真正算满，因为没有那些记录可能达不到最小箱大小；其次，更重要的是，如果问题 FlowFile 在问题记录之前包含 100 条"好"记录，这 100 条记录已经写入了 "merged" FlowFile，我们无法撤销已写入的记录。如果把这 100 条记录继续发送出去，并把问题 FlowFile 路由到 'failure'，那么一旦 "failure" Relationship 最终又被路由回 MergeRecord，就可能不断重复这 100 条已成功处理的记录。

示例

为了更好地理解该 Processor 的工作方式，我们给出几个示例。为简单起见，这些示例使用 CSV 格式的数据并把合并结果也写为 CSV 格式，但数据格式本身并不重要，只要有能读取数据的 Record Reader 和能以期望格式写出数据的 Record Writer 即可。

示例 1 - 把许多小 FlowFile 组成批次

当我们想把许多小 FlowFile 组成批次以创建一个更大的 FlowFile 时，可以通过 "Bin Packing Algorithm" 合并策略实现。思路是在最小/最大记录数和箱大小的范围内， 尽可能多地把 FlowFile 打包在一起。假设设置了以下属性：

再假设队列中有如下数据，schema 包含 Name 和 Age 两个字段：

由于没有配置 Correlation Attribute，且所有 FlowFile 具有相同的 schema，Processor 会 尝试把这些 FlowFile 全部加入同一个箱。因为 Minimum Number of Records 是 3、 Maximum Number of Records 是 5，所有 FlowFile 都会被加入同一个箱。输出是一个内容为 如下所示的单个 FlowFile：

Mark, 33 John, 45 Jane, 43 Jake, 3 Jan, 2

Processor 运行时，会把能从队列取到的 FlowFile 都装箱。之后，它会合并任何"足够满"的箱。 因此，如果队列中只有 3 个 FlowFile，这 3 个会被加入，等第 4 个出现时会在下一次迭代中 创建新箱。但如果队列中有 8 个，只有 5 个会被加入第一个箱，另外 3 个会被加入第二个箱， 该箱因为也达到了最小阈值 3 而会被合并。