Kafka Streams 按复杂条件键加入 [英] Kafka Streams join by key with complex condition

问题描述

我正在尝试通过键将 KStream 与 GlobalKTable 连接起来，但具有特定的逻辑.

I'm trying to join KStream with GlobalKTable by key, but with specific logic.

    StreamsBuilder builder = new StreamsBuilder();
    KStream<String, Integer> stream = builder.stream(inputTopic1); // key = "ABC"
    GlobalKTable<String, Integer> table = builder.globalTable(inputTopic2); // key = "ABC"

    stream.join(table, // join first by "ABC" = "ABC", then by "AB" = "AB", then by "A" = "A"
            (key, value) -> key,
            (valueLeft, valueRigth) -> {/* identify by which condition the join was performed */});

例如，如果key = "ABC"，则:

For example, if the key = "ABC", then:

• 首先，通过完整键加入 - 即ABC"=ABC"
• 然后，如果未连接，则通过前两个符号连接(删除一个符号) - 即AB"=AB"
• 最后，尝试只加入一个符号 - 即A"=A"

此外，还需要知道连接是在哪个条件下执行的 - 例如，通过 3 个字母/通过 2 个字母/通过 1 个字母.

Additionally, it is required to know by which condition was the join performed - e.g., by 3 letters / by 2 letters / by 1 letter.

问题是，是否有可能，或者我应该寻找解决方法?例如，使用相应的键(带有ABC"键的表，带有AB"键的表和带有A"键的表)制作 GlobalKTable 的副本并执行 3 个单独的连接?或者有什么其他建议?

The question is, is it possible at all or should I search for a workaround? For example, make copies of GlobalKTable with corresponding keys (table with "ABC" key, one with "AB" key and one with "A" key) and perform 3 separate joins? Or maybe any other suggestions?

提前致谢！

推荐答案

对多个表使用一系列左联接是可能的(如果您知道经常想要尝试联接).如果连接成功，则跳过下一个连接.使用 leftJoin() 和 branch() 的组合应该允许您在每次加入后将流拆分为加入"和重试".最后，如果需要，您可以 merge() 将不同的结果流放在一起.

Using a series of left-joins against multiple tables would be possible (if you know of often you want to try the join). If the joins succeeds, you skip the next join. Using a combination of leftJoin() and branch() should allow you split the stream after each join into "joined" and "retry". At the end, you can merge() the different result streams together if you want.

这篇关于Kafka Streams 按复杂条件键加入的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！