Kafka Streams接口: KStream to KTable

Question

我有一个Kafka主题，我在其中发送定位事件(key=user_id，value=user_location)。我能够将其作为KStream读取和处理

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

这很好用，但我想要一个包含每个用户最近已知位置的KTable。我怎么能做到呢？

我可以通过写入和读取中间主题来实现：

// 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

更新：

在Kafka 2.5中，将添加一个新的方法KStream#toTable()，这将提供一种将KStream转换为KTable的便捷方法。有关详情，请参阅：https://cwiki.apache.org/confluence/display/KAFKA/KIP-523%3A+Add+KStream%23toTable+to+the+Streams+DSL

原始答案：

目前还没有直接的方法来做到这一点。您的方法是绝对有效的，如Confluent FAQ中所述：http://docs.confluent.io/current/streams/faq.html#how-can-i-convert-a-kstream-to-a-ktable-without-an-aggregation-step

这是关于代码的最简单的方法。然而，它有以下缺点：(a)您需要管理额外的主题；(b)由于数据被写入Kafka和从Kafka重新读取，它会导致额外的网络流量。

还有一种选择，使用"dummy-reduce"：

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中的手动主题管理之间的权衡。