如何创建单个Kafka Stream实现从INPUT_TOPIC读取并写入OUTPUT_TOPIC？

完全可行，你可以通过单个KStream完成从INPUT_TOPIC读取指定字段并写入OUTPUT_TOPIC的操作，无需拆分两个流，这样既简化结构又能保证功能和原有实现一致。

针对KSQL场景 #

如果使用KSQL，原本的双流写法可以合并为单条语句：

-- 合并后的单流写法
CREATE STREAM OUTPUT_STREAM WITH (KAFKA_TOPIC='OUTPUT_TOPIC', VALUE_FORMAT='JSON') AS
SELECT timestamp, DIMENSION
FROM INPUT_TOPIC;

如果需要显式指定源主题的字段结构，也可以写成：

CREATE STREAM OUTPUT_STREAM AS
SELECT timestamp, DIMENSION
FROM INPUT_TOPIC (VALUE_FORMAT='JSON', KEY_FORMAT='STRING')
INTO OUTPUT_TOPIC;

针对Java Kafka Streams API场景 #

如果使用Java API，原本的拆分流写法可以合并为链式调用：

// 合并后的单流代码
StreamsBuilder builder = new StreamsBuilder();

builder.stream("INPUT_TOPIC", Consumed.with(Serdes.String(), JsonSerdes.JsonNode()))
    // 提取需要的字段
    .mapValues(value -> {
        ObjectNode result = JsonNodeFactory.instance.objectNode();
        result.put("timestamp", value.get("timestamp").asLong());
        result.put("DIMENSION", value.get("DIMENSION").asText());
        return result;
    })
    // 直接写入目标主题
    .to("OUTPUT_TOPIC", Produced.with(Serdes.String(), JsonSerdes.JsonNode()));

KafkaStreams streams = new KafkaStreams(builder.build(), new StreamsConfig(props));
streams.start();

这种合并方式减少了中间流的冗余处理，性能上更高效，只要保证读取的字段在INPUT_TOPIC消息中存在，且序列化/反序列化配置正确，就能和原有双流实现完全等价。

内容的提问来源于stack exchange，提问作者alphanumeric