Apache Flink RocksDB状态管理

Question

我在同一份flink jobs中读了两个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. 可以Streu2访问由Streu1保存的相同memberId的状态吗？

Answer 1

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

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

通过这种方式，您可以将状态保持在单个键控协处理函数中，这样您就可以访问stream1和stream2的状态。

因此，对于processElement1您可以在map中为stream1执行相同的操作，而在processElement2中，您可以在map中为stream2执行相同的操作。