在Flink中缓存外部数据库的最佳方式是什么？

Question

外部数据库由每个键的一组规则组成，这些规则应该应用于Flink作业中的每个流元素。因为为每个元素调用DB并检索规则的开销非常大，所以我希望在初始化时从数据库获取规则，并将其存储在本地缓存中。

当在外部数据库中更新规则时，将向Flink作业发布一个状态更改事件，该事件应用于获取规则并刷新此缓存。

实现我所描述的目标的最佳方法是什么？我查看了键控状态，但初始化所有键并在更新时刷新键似乎是不可能的。

Answer 1

我认为你可以利用BroadcastProcessFunction or KeyedBroadcastProcessFunction来实现你的用例。一个详细的博客，可访问here

简而言之，：您可以定义源，如Kafka或任何其他源，然后将您希望实际流使用的规则发布到Kafka。连接实际的数据流和规则流。然后，processBroadcastElement将流式传输规则，您可以在其中更新状态。最后，可以在实际的事件流方法processElement中检索更新后的状态(规则)。

需要考虑的几点：广播状态将始终保存在堆中，而不是状态存储(RocksDB)中。因此，它必须足够小才能放入内存中。每个插槽都会将所有广播状态拷贝到其检查点，因此所有检查点和保存点都将拥有n个(并行度)广播状态拷贝。

Answer 2

Flink中的一些不同机制可能与此用例相关，具体取决于您的详细需求。

广播状态

Jaya Ananthram已经介绍了在his answer中使用广播状态的想法。如果规则应该全局应用于每个键，并且如果您可以找到收集和广播更新的方法，则这是有意义的。

请注意，KeyedBroadcastProcessFunction方法的processBroadcastElement()中的Context包含方法applyToKeyedState(StateDescriptor<S, VS> stateDescriptor, KeyedStateFunction<KS, S> function)。这意味着您可以注册一个将应用于与所提供的stateDescriptor关联的所有密钥的所有状态的KeyedStateFunction。

状态处理器应用编程接口

如果您希望从数据库转储中引导Flink保存点中的状态，则可以使用此库执行此操作。您将找到一个使用State Processor API在this gist中引导状态的简单示例。

变更数据捕获

表/SQL支持Debezium、Canal、Maxwell CDC streams和Kafka upsert streams。这可能是一种解决方案。还有flink-cdc-connectors。

加入的查找

Flink SQL可以使用configurable cache对JDBC数据库执行临时查找联接。不确定这是否相关。

Answer 3

从本质上说，David的回答很好地概括了这一点。2

1

2