如何在Apache Flink中从数据库中查找和更新记录的状态？

Question

理想情况下，与特定记录相关的事件需要以FIFO顺序处理（尽管也会有一个时间戳来帮助检测无序事件），但与不同记录相关的事件可以并行处理。我计划使用keyby()构造按记录对流进行分区。

需要进行的处理取决于数据库中关于记录的当前信息。但是，我找不到一个示例或推荐的方法来查询数据库中的记录，以丰富事件，它正在被处理，我需要处理它的附加信息。

我想到的管道如下：

->接收到的id上的keyBy()->从对应于id的数据库中检索记录->对记录执行处理步骤->将处理过的事件推送到外部队列并更新数据库记录

Answer 1

您可以通过为RichFlatMap函数扩展rich函数来执行数据库查找，在其Open()方法中初始化数据库连接一次，然后在FlatMap()方法中处理每个事件：

public static class DatabaseMapper extends RichFlatMapFunction<Event, EncrichedEvent> {

    // Declare DB coonection and query statements

    @Override
    public void open(Configuration parameters) throws Exception {
      // Initialize Database connection
      // Prepare Query statements
    }

    @Override
    public void flatMap(Event currentEvent, Collector<EncrichedEvent> out) throws Exception {
      // look up the Database, update record, enrich event
      out.collect(enrichedEvent);        
    }
})

然后可以使用databaseMapper，如下所示：

stream.keyby(id)
      .flatmap(new DatabaseMapper())
      .addSink(..);

您可以在这里找到一个使用来自Redis的缓存数据的示例。