在创建它的同一个应用程序中查询 KTable答案

【问题标题】：Query KTable in the same Application where it is created在创建它的同一个应用程序中查询 KTable
【发布时间】：2020-10-22 11:26:57
【问题描述】：

我有一个 Kafka 流应用程序，我在其中读取一个主题，进行聚合并在 KTable 中实现。然后我创建一个 Stream 并在流上运行一些逻辑。现在在流处理中，我想使用前面提到的 KTable 中的一些数据。启动流应用程序后，如何再次访问 KTable 流？我不想将 KTable 推送到新主题。

KStream<String, MyClass> source = builder.stream("my-topic");
KTable<Windowed<String>, Long> kTable =
            source.groupBy((key, value) -> value.getKey(),
                    Grouped.<String, MyClass >as("repartition-1")
                            .withKeySerde(new Serdes.String())
                            .withValueSerde(new MyClassSerDes()))
                    .windowedBy(TimeWindows.of(Duration.ofSeconds(5)))
                    .count(Materialized.<String, Long, WindowStore<Bytes, byte[]>>as("test-store")
                            .withKeySerde(new Serdes.String())
                            .withValueSerde(Serdes.Long()));

这里我想使用 kTable 中的数据。

inputstream.groupByKey()
    .windowedBy(TimeWindows.of(Duration.ofMinutes(1)))
    .count(Materialized.<myKey, Long, WindowStore<Bytes,   byte[]>>as("str")
    .withRetention(Duration.ofMinutes(30)))
    .toStream()
    .filter((k, v) -> { 
        // Here get the count for the previous Window.
        // Use that count for some computation here.
    }

【问题讨论】：

如果取决于您到底想做什么...也许您可以使用transform() 并将KTable 存储连接到变压器？
感谢您的回复。在我的 Streams 应用程序中，通过对 Stream 进行一些聚合，我通过一个键创建了 Windowed 计数的 KTable。在同一个 Kafka 流应用程序中，我想访问上一个 TimeWindow 的计数。这在不需要额外的中间主题的情况下是否可行？
应该是可行的——你需要什么确切的访问权限？如前所述，您可以将KTable 状态连接到transform() 以获得对其的读取访问权限。
@MatthiasJ.Sax 更新了问题。请检查标有“这里我想使用 kTable 中的数据”的部分。
您可以将filter() 替换为flatTransformValues(..., "my-key-count-one-minute") 并在您的转换器中实现过滤器，并且可以通过上下文访问状态：docs.confluent.io/current/streams/developer-guide/…

标签： apache-kafka-streams ktable

【解决方案1】：

您可以将KTable 存储添加到处理器/变压器。对于您的情况，您可以将 filter 替换为 flatTransform（或任何兄弟，如 transform 等，具体取决于您是否需要访问密钥）并将商店连接到运营商：

inputstream.groupByKey()
    .windowedBy(TimeWindows.of(Duration.ofMinutes(1)))
    .count(Materialized.<myKey, Long, WindowStore<Bytes, byte[]>>as("str")
        .withRetention(Duration.ofMinutes(30))
    )
    .toStream()
    // requires v2.2; otherwise use `transform()`
    // if you don't need access to the key, consider to use `flatTransformValues` (v2.3)
    .flatTransform(
        () -> new Transformer<Windowed<myKey>,
                              Long,
                              List<KeyValue<Windowed<myKey>, Long>>() {

            private ReadOnlyWindowStore<myKey, Long> store;

            public void init(final ProcessorContext context) {
                // get a handle on the store by its name
                // as specified via `Materialized` above;
                // should be read-only
                store = (ReadOnlyWindowStore<myKey, Long>)context.getStateStore("str");
            }

            public List<KeyValue<Windowed<myKey>, Long>> transform(Windowed<myKey> key,
                                                                   Long value) {

              // access `store` as you wish to make a filtering decision

              if ( ... ) {
                  // record passes
                  return Collection.singletonList(KeyValue.pair(key, value));
              } else {
                  // drop record
                  return Collection.emptyList();
              }
            }

            public void close() {} // nothing to do
        },
        "str" // connect the KTable store to the transformer using its name
              // as specified via `Materialized` above
    );

【讨论】：