KStream-GlobalKTable join 可以为特定搜索返回多个匹配记录吗？

Question

我希望有人可以帮助解决我在 Kafka 中遇到的有关 GlobalKTables 的问题。

我正在尝试执行 KStream-GlobalKTable 连接。但是，我想检索 GlobalKTable 中其键 or 值包含在我的原始流事件中找到的字符串的所有条目。例如，假设我的表有 3 行，其中包含以下键：

Key: BANK055DEPOSIT value: {some data}
 Key: BANK055CREDIT value: {different data}
 Key: BANK033CREDIT value: {more different data}

当我对表进行连接以检索数据时，我需要拉回所有键或值包含“055”的行。所以我想要前两行。

在数据库世界中，这相当于以下内容：

SELECT * FROM GlobalKTable where table_key.contains("055”) OR table_value.contains(“055”)

我搜索了官方文档，但没有找到如何执行此操作的示例。我怀疑无法从 GlobalKTable 连接中检索 N 行。

另外，我为此使用了流 DSL。不确定这是否可以使用处理器 API 实现。 任何意见表示赞赏！

Answer 1

当joining a KStream with a GlobalKTable 时，您可以使用KStream, 的部分键和值，但它最终必须匹配整个GlobalKTable 键，因此，很遗憾，您不能使用上面所说的加入。

但是即使使用 DSL，您仍然应该能够做一些接近此的事情。如果您使用KStream.transformValues with a ValueTransformerWithKeySupplier，您可以扫描状态存储并根据流记录中包含的子字符串提取您想要的记录。此外，您不必扫描整个商店，而是使用range query。

编辑：这是我正在努力展示的一些代码。

@SuppressWarnings("unchecked")
public class MultiResultJoinExample {

    public static void main(String[] args) {
        final Properties props = new Properties();
        props.put(StreamsConfig.APPLICATION_ID_CONFIG, "mult-partial-key-join-results");
        props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        final StreamsBuilder builder = new StreamsBuilder();

        final String storeName = "kv-store";
        final StoreBuilder<KeyValueStore<String, String>> keyValueStoreBuilder =
                Stores.keyValueStoreBuilder(Stores.persistentKeyValueStore(storeName),
                        Serdes.String(),
                        Serdes.String());
        builder.addStateStore(keyValueStoreBuilder);

        final KStream<String, String> streamToJoinAgainst = builder.stream("to-join-input", Consumed.with(Serdes.String(), Serdes.String() ));

        streamToJoinAgainst.transformValues(new StoringValueTransformer(storeName), storeName);

        final KStream<String, String> streamNeedingJoin = builder.stream("need-join-input", Consumed.with(Serdes.String(), Serdes.String()));

        streamNeedingJoin.flatTransformValues(new FlatMapJoiningTransformer(storeName), storeName).to("output", Produced.with(Serdes.String(), Serdes.String()));

        final KafkaStreams streams = new KafkaStreams(builder.build(props), props);
        streams.start();
    }

    static final class FlatMapJoiningTransformer implements ValueTransformerWithKeySupplier<String, String, Iterable<String>> {
        final String storeName;

        public FlatMapJoiningTransformer(String storeName) {
            this.storeName = storeName;
        }

        @Override
        public ValueTransformerWithKey<String, String, Iterable<String>> get() {
            return new ValueTransformerWithKey<String, String, Iterable<String>>() {
               private KeyValueStore<String, String> kvStore;
                @Override
                public void init(ProcessorContext<Void, Void> context) {
                    kvStore = (KeyValueStore<String, String>) context.getStateStore(storeName);
                }

                @Override
                public Iterable<String> transform(String readOnlyKey, String value) {
                      List<String> results = new ArrayList<>();
                      final String patternToMatch = readOnlyKey.substring(4, 7);
                      try (KeyValueIterator<String, String> iter =  kvStore.all()) {
                           while(iter.hasNext()) {
                               final KeyValue<String, String> kv = iter.next();
                                if (kv.key.contains(patternToMatch) || kv.value.contains(patternToMatch)){
                                    results.add(kv.value + " - " + value);
                                }
                           }
                      }
                      return results;
                }

                @Override
                public void close() {

                }
            };
        }
    }

    static final class StoringValueTransformer implements ValueTransformerWithKeySupplier<String, String, String> {
        final String storeName;

        public StoringValueTransformer(String storeName) {
            this.storeName = storeName;
        }

        @Override
        public ValueTransformerWithKey<String, String, String> get() {
            return new ValueTransformerWithKey<String, String, String>(){
                private KeyValueStore<String, String> kvStore;
                @Override
                public void init(ProcessorContext<Void, Void> context) {
                       kvStore = (KeyValueStore<String, String>)context.getStateStore(storeName);
                }

                @Override
                public String transform(String readOnlyKey, String value) {
                    kvStore.putIfAbsent(readOnlyKey, value);
                    return value;
                }

                @Override
                public void close() {
                     //no-op
                }
            };
        }
    }
}

HTH, 比尔