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 */});

例如，如果键="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()应该可以让您在每次联接后将流分割为"joined"和"retry".最后，如果需要，您可以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屋！