如何在Kafka流应用程序中迭代GlobalKTable或KTable

Question

我有一个Kafka流应用程序，有两个数据源：事件和用户。

我有4个主题：事件，用户，用户2，用户事件

Users2与Users相同，用于演示GlobalKTable。

Events主题使用时间戳模式，因此当到达时间戳字段日期时，KStream将接收事件记录。

此时，我想为用户KTable中的每个用户ID以及新的事件ID创建一个用户事件记录；但我不知道如何遍历GlobalKTable或KTable来实现这一点。

public Topology createTopology() {

    final Serde<String> serde = Serdes.String();
    final StreamsBuilder builder = new StreamsBuilder();

    final GlobalKTable<String, String> gktUsers =
            builder.globalTable("Users",
                    Materialized.<String, String, KeyValueStore<Bytes, byte[]>>as("user-store")
                            .withKeySerde(Serdes.String()).withValueSerde(serde));

    final KTable<String, String> ktUsers = builder.table("Users2");

    builder.stream("Events", Consumed.with(Serdes.String(), serde))
            .peek((k, v) -> {
                // This is called when a new Event record becomes current.
                // How do I iterate through gktUsers at this point
                // and output a User-ID and an Event-ID to the User-Events topic?

                //  This type of iteration doesn't work either.
                ktUsers.toStream().foreach(new ForeachAction<String, String>() {
                    @Override
                    public void apply(String s, String s2) {
                        log.info("{} {}", s, s2);
                    }
                });
            })
            
    builder.build();
}

Answer 1

你需要让状态存储迭代它

ReadOnlyKeyValueStore<String, String> keyValueStore = streams.store("user-store", QueryableStoreTypes.keyValueStore()));

String value = store.get("key");

KeyValueIterator<String, String> range = keyValueStore.all(); 

https://docs.confluent.io/platform/current/streams/developer-guide/interactive-queries.html

或者，也许您应该在用户和事件之间进行流表连接