Kafka Streams API:KStream到KTable

我有一个Kafka主题,我发送位置事件(key = user_id,value = user_location).我能够阅读并处理它KStream:

KStreamBuilder builder = new KStreamBuilder();

KStream 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);
        });

这很有效,但我想拥有KTable每个用户的最后一个已知位置.我怎么能这样做？

我能够写一个中间主题并从中读取:

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

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

有没有一种简单的方法来获得KTable一个KStream？这是我第一个使用Kafka Streams的应用程序,所以我可能会遗漏一些明显的东西.

目前没有直接的方法来做到这一点.您的方法绝对有效,如汇总常见问题解答中所述:http://docs.confluent.io/current/streams/faq.html#how-can-i-convert-a-kstream-to-a-ktable-without-an -aggregation步

这是关于代码的最简单方法.但是,它的缺点是:(a)您需要管理其他主题,并且(b)由于数据被写入Kafka并从Kafka重新读取,因此会导致额外的网络流量.

有一种替代方案,使用"虚拟减少":

KStreamBuilder builder = new KStreamBuilder();
KStream stream = ...; // some computation that creates the derived KStream

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