在 Flink 中用 (X) 键控的缓慢变化流来丰富 (X,Y) 键控的快速流

Posted 2023-04-18

【中文标题】在 Flink 中用 (X) 键控的缓慢变化流来丰富 (X,Y) 键控的快速流

【英文标题】：Enrich fast stream keyed by (X,Y) with a slowly change stream keyed by (X) in Flink

【发布时间】：2019-11-27 15:16:11

【问题描述】：

我需要通过（userId）键控的缓慢变化的streamB 来丰富我快速变化的streamA（用户ID，startTripTimestamp）键控。

我使用 Flink 1.8 和 DataStream API。我考虑了两种方法：

广播streamB 并通过 userId 和最近的时间戳加入流。它会等同于 TableAPI 中的 DynamicTable 吗？我可以看到这个解决方案的一些缺点：streamB 需要适合每个工作节点的 RAM，它增加了 RAM 的利用率，因为整个 streamB 需要存储在每个工作节点的 RAM 中。

将streamA 的状态概括为仅由 (userId) 键入的流，我们将其命名为 streamC，以便与 streamB 具有共同键。然后我可以将streamC 与streamB 合并，按处理时间排序，并在状态下处理这两种类型的事件。处理生成的流（处理函数中的更多代码）更复杂，但不会消耗那么多 RAM 以在所有节点上拥有所有 streamB。此解决方案是否还有其他缺点或优点？

我也看到了这个提议https://cwiki.apache.org/confluence/display/FLINK/FLIP-17+Side+Inputs+for+DataStream+API 里面说的：

一般来说，大部分都遵循加入主流的模式 具有一个或多个缓慢变化的输入的高吞吐量或 静态数据：

[...]

使用缓慢演变的数据加入流：这非常类似于 上述情况，但我们用于丰富的侧面输入是 随着时间的推移而演变。这可以通过等待一些初始数据来完成 在处理主要输入和连续 将新数据摄取到内部输入结构中 到了。

不幸的是，要实现此功能https://issues.apache.org/jira/browse/FLINK-6131 似乎还需要很长时间，并且没有描述替代方案。因此，我想询问当前针对所描述的用例推荐的方法。

我见过Combining low-latency streams with multiple meta-data streams in Flink (enrichment)，但它没有指定该流的键是什么，而且它是在 Flink 1.4 时回答的，所以我预计推荐的解决方案可能已经改变。

【参考方案1】：

以 Gaurav Kumar 已经回答的内容为基础。

主要问题是您需要精确匹配来自streamA 和streamB 的记录还是尽力匹配？例如，由于竞争条件，来自streamA 的一些（很多？）记录可以在来自streamB 的一些更新到达之前处理，例如在启动期间，这对您来说是个问题吗？

我建议从Table API is solving this issue 中汲取灵感。可能 Temporal Table Join 是您的正确选择，这会让您做出选择：处理时间还是事件时间？

Gaurav Kumar 的两个提案都是 of processing time Temporal Table joins 的实现，它假设记录可以非常松散地连接，并且不必正确计时。

如果来自streamA 和streamB 的记录必须正确计时，那么您必须以一种或另一种方式缓冲两个流中的一些记录。有多种方法可以做到这一点，具体取决于您想要实现的semantic。决定了之后，实际的实现就没有那么难了，你可以从 Table API join 操作符（flink-table-planner 模块中的org.apache.flink.table.runtime.join 包）中汲取灵感。

侧输入（您引用的）和/或输入选择只是用于控制不必要的缓冲记录数量的工具。您可以在没有它们的情况下实现有效的 Flink 作业，但如果一个流明显超过另一个流，则内存消耗可能难以控制（就事件时间而言 - 对于处理时间而言，这不是问题）。

【讨论】：

我想补充一点，如果可能，请尝试使用 Table API。您可以转换为data stream back and forth to a table。

【参考方案2】：

答案取决于streamB 状态的大小，需要使用它来丰富streamA

如果您广播您的streamB 状态，那么您将把streamB 中的所有用户ID 放到每个任务管理器中。任务管理器上的每个任务都只会有来自 streamA 的这些 userId 的子集。因此，来自 streamB 的一些 userId 数据将永远不会被使用，并将作为一种浪费。因此，如果您认为streamB 状态的大小不足以真正影响您的工作，并且不会占用大量内存来为状态管理留下更少的内存，您可以保留整个streamB 状态。这是您的第一名。 如果您的streamB 状态真的很大并且会消耗任务管理器的大量内存，您应该考虑方法#2。两个流的 KeyBy 相同 ID 以确保具有相同用户 ID 的元素到达相同的任务，然后您可以使用托管状态来维护每个键的 streamB 状态并使用此托管状态丰富 streamA 元素。