Apache Flink RocksDB 状态管理

Question

我在同一个 flink 作业中阅读 2 个 kafka 主题。

• Stream1: 来自第一个主题的消息保存到rocksdb，然后与stream2联合。
• Stream2: 来自第二个主题的消息被 stream1 保存的状态丰富了，然后它将与 stream1 合并。

主题 1 和主题 2 是不同的来源，但基本上两个来源的输出是相同的。我必须使用来自 topic1 的数据来丰富来自 topic2 的数据。

这里是流程；

val stream1 = readKafkaTopic1().keyBy(_.memberId).map(saveMemberDetailsToRocksDB)
val stream2 = readKafkaTopic2().keyBy(_.memberId).map(readMemberDetailsAndEnrich)
stream1.union(stream2).addSink(kafkaProducer)

这是问题；

1. 流动性好吗？
2. stream2可以访问stream1保存的相同memberId的状态吗？

Answer 1

似乎您应该能够通过使用KeyedCoProcessFunction 来实现您想要的。这或多或少会这样：

stream1
.keyBy(_.memberId)
.connect(stream2.keyBy(_.memberId))
.process(new CustomKeyedCoProcessFunction())

这样，您可以将状态保持在单个 KeyedCoProcessFunction 中，因此您可以同时访问 stream1 和 stream2。

所以，对于processElement1，您可以在map 中为stream1 和在processElement2 中做同样的事情您可以在map 中为stream2 做同样的事情。