Kafka Streams API：KStream 到 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 常见问题解答中所述，您的方法绝对有效：http://docs.confluent.io/current/streams/faq.html#how-can-i-convert-a-kstream-to-a-ktable-without-an-aggregation-step

这是最简单的代码方法。但是，它的缺点是（a）您需要管理一个额外的主题，并且（b）它会导致额外的网络流量，因为数据是从 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 将创建一个内部更改日志主题来备份 KTable 以实现容错。因此，这两种方法都需要 Kafka 中的一些额外存储，并导致额外的网络流量。总体而言，这是选项 2 中稍微复杂的代码与选项 1 中手动主题管理之间的权衡。