如何基于条件实时拆分Kafka Connect同步的ES数据主题

核心思路 #

直接用Kafka Streams做流式处理：从ES同步过来的源主题消费消息，解析payload里的description字段，匹配关键词后转发到对应目标主题。相比Python消费，Streams原生基于Kafka的消费/生产机制，自带容错、负载均衡和低延迟，完全满足实时性要求。

具体实现步骤（Java版，Kafka Streams原生支持最优） #

1. 基础配置 #

先搭好Streams应用的核心配置，指定Kafka地址、应用ID（用来做状态管理）和默认的序列化/反序列化器：

import org.apache.kafka.streams.StreamsConfig;
import org.apache.kafka.common.serialization.Serdes;
import java.util.Properties;

Properties props = new Properties();
props.put(StreamsConfig.APPLICATION_ID_CONFIG, "es-data-splitter");
props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "your-kafka-broker:9092");
props.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass());
props.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.String().getClass());

2. 读取源主题并解析Payload #

假设同步过来的消息是JSON格式，用Jackson解析payload，提取description字段：

import org.apache.kafka.streams.StreamsBuilder;
import org.apache.kafka.streams.kstream.KStream;
import com.fasterxml.jackson.databind.ObjectMapper;
import java.util.Map;

StreamsBuilder builder = new StreamsBuilder();
// 替换成你的ES同步源主题
KStream<String, String> sourceStream = builder.stream("prefix_your_es_index");

ObjectMapper jsonMapper = new ObjectMapper();

3. 关键词匹配与主题转发 #

分两种场景处理：

场景1：一个消息只匹配一个关键词（互斥转发）

用branch方法把流拆分成多个分支，每个分支对应一个关键词，最后分别转发到目标主题：

import org.apache.kafka.streams.kstream.Predicate;

// 定义每个关键词的匹配规则
Predicate<String, String> matchEnergy = (key, value) -> {
    try {
        Map<String, Object> payload = jsonMapper.readValue(value, Map.class);
        String desc = (String) payload.get("description");
        return desc != null && desc.contains("energy");
    } catch (Exception e) {
        return false;
    }
};

Predicate<String, String> matchElectric = (key, value) -> {
    try {
        Map<String, Object> payload = jsonMapper.readValue(value, Map.class);
        String desc = (String) payload.get("description");
        return desc != null && desc.contains("electric");
    } catch (Exception e) {
        return false;
    }
};

Predicate<String, String> matchTemperature = (key, value) -> {
    try {
        Map<String, Object> payload = jsonMapper.readValue(value, Map.class);
        String desc = (String) payload.get("description");
        return desc != null && desc.contains("temperature");
    } catch (Exception e) {
        return false;
    }
};

// 分支：最后一个Predicate是匹配所有未命中的消息
KStream<String, String>[] streamBranches = sourceStream.branch(
    matchEnergy,
    matchElectric,
    matchTemperature,
    (key, value) -> true
);

// 每个分支转发到对应主题
streamBranches[0].to("topic_energy");
streamBranches[1].to("topic_electric");
streamBranches[2].to("topic_temperature");
// 未命中的消息转发到备用主题
streamBranches[3].to("topic_unmatched");

场景2：一个消息可能匹配多个关键词（多主题转发）

如果一条消息的description同时包含多个关键词（比如既提到energy又提到electric），需要同时转发到多个主题，用foreach遍历处理：

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;
import java.util.Properties;

// 配置Producer（也可以复用Streams的配置）
Properties producerProps = new Properties();
producerProps.put("bootstrap.servers", "your-kafka-broker:9092");
producerProps.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
producerProps.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
KafkaProducer<String, String> producer = new KafkaProducer<>(producerProps);

sourceStream.foreach((key, value) -> {
    try {
        Map<String, Object> payload = jsonMapper.readValue(value, Map.class);
        String desc = (String) payload.get("description");
        if (desc == null) {
            producer.send(new ProducerRecord<>("topic_unmatched", key, value));
            return;
        }
        // 匹配一个就发送一个
        if (desc.contains("energy")) {
            producer.send(new ProducerRecord<>("topic_energy", key, value));
        }
        if (desc.contains("electric")) {
            producer.send(new ProducerRecord<>("topic_electric", key, value));
        }
        if (desc.contains("temperature")) {
            producer.send(new ProducerRecord<>("topic_temperature", key, value));
        }
        // 如果没有匹配任何关键词，转发到备用主题
        if (!desc.contains("energy") && !desc.contains("electric") && !desc.contains("temperature")) {
            producer.send(new ProducerRecord<>("topic_unmatched", key, value));
        }
    } catch (Exception e) {
        // 解析失败的消息发送到死信队列
        producer.send(new ProducerRecord<>("topic_dead_letter", key, value));
        e.printStackTrace();
    }
});

4. 错误处理 #

一定要处理JSON解析失败的情况，把这类消息转发到死信队列（topic_dead_letter），避免丢消息，后续还可以手动排查解析问题。

部署运行 #

把代码打包成jar包，用Kafka官方提供的kafka-streams-application.sh脚本启动，或者直接用java -jar your-app.jar运行。Streams会自动加入消费者组，做负载均衡和故障转移。

为什么不选Python消费？ #

Python的Kafka客户端虽然能实现消费转发，但需要自己处理偏移量提交、多线程负载、故障恢复等逻辑，实时性和可靠性远不如Kafka Streams——Streams是原生流式处理框架，延迟在毫秒级，而且自带状态管理和容错机制，不用自己造轮子。

内容的提问来源于stack exchange，提问作者Sarindra Thérèse