Flink 如何处理 Union 算子的水印？

Question

我从四个 Kinesis 流中读取数据。每个流中的数据是不同的数据类型。在读入所有四个流之后，我分配时间戳和水印，并聚合来自每个流的数据。四个聚合的结果都使用相同的通用对象输出。我想合并来自四个流的结果，以便可以将合并的流发送到一个 ProcessFunction。这基本上允许我像 CoProcessFunction 一样使用 ProcessFunction，但我将能够处理来自两个以上流的数据（在这种情况下，ProcessFunction 将接收来自所有四个单独流的聚合）。

但是，我担心这可能无法很好地处理水印。如果一个流需要更长的时间来处理或以某种方式落后，如果所有水印在联合中向前传递并且其中一个流领先于其他流，则它的聚合可能无法进入处理函数。如果是这种情况，那么处理函数的水印将是它从四个单独的流中看到的水印的最大值。

我的问题是：联合运营商如何处理水印，联合运营商下游的运营商如何处理这些水印？

另外：如果泛型对象的联合由于水印问题而不起作用，当 Flink 仅支持两个流的 CoProcessFunction 时，将四个不同聚合的结果组合起来的最佳方法是什么？

Answer 1

将两个以上的流连接在一起的另一种方法是构建一棵树，该树进行成对连接，直到所有流都连接在一起。要么作为平衡树，像这样：

A--->
     A+B---->
B--->

            A+B+C+D------------>

C--->
     C+D---->
D--->

或一次添加一个流，如下所示：

a--->
     a+b--->
b--->
            a+b+c--->
     c----->
                     a+b+c+d--->
            d------->

FWIW，FLIP-92 是向 Flink 添加 n 元流运算符的提议，但即使实施，它也可能不会对用户可见，至少一开始是这样。

Answer 2

联合水印的工作原理与并行流的水印一样。这意味着水印始终是来自所有输入流的水印的min。同样代表下游操作符，它们的水印将是所有输入流的min。

老实说，我认为联合并不依赖于水印。但是，如果您出于任何原因想要使用 CoProcessFunction，我可以提供这种有点 hacky 的方式。您可以创建一个 Seq 您已生成的流，然后：

//Streams defined
val seq = Seq(stream, stream2, stream3, stream4)
seq.reduce((stream1, stream2) => stream1.connect(stream2).process(...))