Kafka Streams API：KStream到KTable

Question

KStreamBuilder builder = new KStreamBuilder();

KStream<String, Location> locations = builder
        .stream("location_topic")
        .map((k, v) -> {
            // some processing here, omitted form clarity
            Location location = new Location(lat, lon);
            return new KeyValue<>(k, location);
        });

我能做到写作和阅读的中间主题：

// write to intermediate topic
locations.to(Serdes.String(), new LocationSerde(), "location_topic_aux");

// build KTable from intermediate topic
KTable<String, Location> table = builder.table("location_topic_aux", "store");

有没有简单的方法从Kstream中获取Ktable？这是我第一个使用Kafka Streams的应用程序，所以我可能错过了一些明显的东西。

Answer 1

更新：

在Kafka2.5中，将添加一个新方法kstream#totable()，它将提供一种方便的方法将kstream转换为ktable。有关详细信息，请参阅:https://cwiki.apache.org/confluence/display/kafka/kip-523%3a+add+kstream%23totable+to+the+streams+DSL

原答案：

KStreamBuilder builder = new KStreamBuilder();
KStream<String, Long> stream = ...; // some computation that creates the derived KStream

KTable<String, Long> table = stream.groupByKey().reduce(
    new Reducer<Long>() {
        @Override
        public Long apply(Long aggValue, Long newValue) {
            return newValue;
        }
    },
    "dummy-aggregation-store");

与选项1相比，这种方法在代码方面稍微复杂一些，但其优点是：(a)不需要手动的主题管理；(b)不需要从Kafka重新读取数据。

总的来说，你需要自己决定，你更喜欢哪种方法：

在选项2中，Kafka Streams将创建一个内部changelog主题来备份KTable以进行容错。因此，这两种方法都需要一些额外的Kafka存储，并导致额外的网络流量。总体而言，这是在选项2中稍微复杂的代码和选项1中的手动主题管理之间的权衡。