kafka 流中 KStreamBuilder.from 的当前方式是什么？答案

【问题标题】：What's the current way of KStreamBuilder.from in kafka stream?kafka 流中 KStreamBuilder.from 的当前方式是什么？
【发布时间】：2016-09-25 13:43:51
【问题描述】：

我正在尝试来自http://kafka.apache.org/0100/javadoc/index.html?org/apache/kafka/streams/KafkaStreams.html的示例代码

但是对于这两行

KStreamBuilder builder = new KStreamBuilder();
builder.from("my-input-topic").mapValue(value -> value.length().toString()).to("my-output-topic");

根据https://kafka.apache.org/0100/javadoc/org/apache/kafka/streams/kstream/KStreamBuilder.html，现在 KStreamBuilder 中似乎没有名为“from”的方法

那么，目前与上述 2 行相同的方法是什么？

Maven 用于管理依赖关系

<dependency>
  <groupId>org.apache.kafka</groupId>
  <artifactId>kafka-streams</artifactId>
  <version>0.10.0.0</version>
</dependency>

【问题讨论】：

标签： apache-kafka

【解决方案1】：

此方法已重命名为 stream()（from() 是旧名称）

KStreamBuilder builder = new KStreamBuilder();
builder.stream("my-input-topic").mapValues(value -> value.length().toString()).to("my-output-topic");

这是文档中的一个错误：https://github.com/apache/kafka/pull/1450

也可以看看这里：http://docs.confluent.io/3.0.0/streams/developer-guide.html#kafka-streams-dsl

【讨论】：

谢谢@Matthias。我想应该是 mapValues 而不是 mapValue，对吧？
是的。文档中还有一个错字。

【解决方案2】：

更重要的是，KStreamBuilder 本身现在已被弃用，而应使用 StreamsBuilder。这不仅仅是更改名称，尽管在大多数情况下，最终用户不会立即需要其他任何东西，apart from the name of the class:

通过 DSL 指定拓扑的两个主要类 (KStreamBuilder) 或处理器 API (TopologyBuilder) 已弃用并由 StreamsBuilder 和 Topology 取代（都是新的类位于包 org.apache.kafka.streams 中）。注意 StreamsBuilder 不扩展拓扑，即类层次结构是现在不同了。新类的方法与旧的通过 DSL 或处理器 API 构建拓扑。然而，有些在 KStreamBuilder 中公开的内部方法和 TopologyBuilder 但不是实际 API 的一部分不存在于不再有新课程。

【讨论】：