如何使用Kafka Streams对Kafka Topic数据执行过滤查询后输出到新Topic

你需要的select * from table1 where col1='a'类条件查询+结果转发到新Topic的需求，属于Kafka Streams的基础场景，用filter算子即可实现，以下是完整实现方案：

1 前置准备 #

首先确认你的源Topic数据的序列化格式，常用的有JSON、Avro、Protobuf，下文以最通用的JSON字符串格式为例，如果你使用结构化schema（如Avro配合Schema Registry），可以直接用对应Serde省去手动解析JSON的步骤。

2 核心实现逻辑 #

整个流程只需要三步：

• 从源Topic读取数据流（对应SQL里的from table1）
• 对每条数据执行条件过滤，只保留col1='a'的记录（对应SQL里的where col1='a'）
• 把过滤后的结果写入目标Topic（对应结果输出）

3 完整Java代码示例 #

import org.apache.kafka.common.serialization.Serdes;
import org.apache.kafka.streams.KafkaStreams;
import org.apache.kafka.streams.StreamsBuilder;
import org.apache.kafka.streams.StreamsConfig;
import org.apache.kafka.streams.kstream.Consumed;
import org.apache.kafka.streams.kstream.Produced;
import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.ObjectMapper;
import java.util.Properties;

public class FilterStreamDemo {
    public static void main(String[] args) {
        // 1 配置Streams基础参数
        Properties props = new Properties();
        props.put(StreamsConfig.APPLICATION_ID_CONFIG, "filter-topic-demo");
        props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "你的kafka地址:9092");
        props.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass());
        props.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.String().getClass());
        // 可选配置：调整处理语义
        props.put(StreamsConfig.PROCESSING_GUARANTEE_CONFIG, StreamsConfig.AT_LEAST_ONCE);

        // 2 构建流处理拓扑
        StreamsBuilder builder = new StreamsBuilder();
        ObjectMapper objectMapper = new ObjectMapper();

        builder.stream("源Topic名称", Consumed.with(Serdes.String(), Serdes.String()))
                // 过滤逻辑，对应where col1='a'，可自行扩展多条件
                .filter((key, value) -> {
                    try {
                        JsonNode jsonNode = objectMapper.readTree(value);
                        return "a".equals(jsonNode.get("col1").asText());
                    } catch (Exception e) {
                        // 解析失败的脏数据直接过滤
                        return false;
                    }
                })
                // 结果写入目标Topic
                .to("目标Topic名称", Produced.with(Serdes.String(), Serdes.String()));

        // 3 启动流应用
        KafkaStreams streams = new KafkaStreams(builder.build(), props);
        streams.start();

        // 优雅关闭钩子
        Runtime.getRuntime().addShutdownHook(new Thread(streams::close));
    }
}

4 关键注意事项 #

• 如果你的数据是Avro/Protobuf格式，直接使用官方对应Serde类，不需要手动解析字符串，过滤时直接读取对象的col1属性判断即可，性能和可靠性更高。
• 如果你需要对相同key的最新数据做查询，才需要使用KTable，纯过滤转发场景用KStream即可，资源开销更低。
• 应用部署后可以通过Kafka自带的命令行工具验证结果，也可以在filter算子中加日志打印排查过滤逻辑是否符合预期。

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