Flink 14.3中KafkaSink如何动态选择写入的Kafka主题

使用KafkaRecordSerializationSchema.builder().setTopic()方式构造的序列化器仅支持写入固定单Topic，无法满足按消息内容路由的需求。Flink 1.14.x版本的KafkaSink原生支持按消息内容动态指定目标Topic，核心是自定义KafkaRecordSerializationSchema接口实现，替换默认的固定Topic序列化器即可。

实现逻辑 #

自定义序列化器时，在每条消息的序列化方法中解析JSON内容，根据配置的路由规则匹配目标Topic，最终返回携带对应Topic信息的ProducerRecord即可，原有KafkaSink的投递保证、服务地址配置等都可以正常保留。

代码实现 #

注意提前初始化JSON解析工具、序列化工具，不要在每条消息处理时重复创建对象避免性能损耗，参考实现如下：

import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.connector.kafka.sink.KafkaRecordSerializationSchema;
import org.apache.flink.connector.kafka.sink.KafkaSinkContext;
import org.apache.kafka.clients.producer.ProducerRecord;
import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.ObjectMapper;
// 其他原有依赖的import

public class DynamicKafkaTopicSerializer implements KafkaRecordSerializationSchema<String> {
    // 全局复用ObjectMapper，避免频繁创建带来的性能损耗
    private static final ObjectMapper OBJECT_MAPPER = new ObjectMapper();
    private final SimpleStringSchema valueSerializer;
    // 路由不匹配/解析失败时的兜底Topic
    private final String defaultTopic;
    // JSON中用来做路由判断的字段名
    private final String routeKey;

    public DynamicKafkaTopicSerializer(String defaultTopic, String routeKey) {
        this.defaultTopic = defaultTopic;
        this.routeKey = routeKey;
        this.valueSerializer = new SimpleStringSchema();
    }

    @Override
    public ProducerRecord<byte[], byte[]> serialize(String element, KafkaSinkContext context, Long timestamp) {
        String targetTopic = defaultTopic;
        try {
            JsonNode jsonNode = OBJECT_MAPPER.readTree(element);
            JsonNode routeValueNode = jsonNode.get(routeKey);
            if (routeValueNode != null) {
                String routeValue = routeValueNode.asText();
                // 按需扩展路由规则，也可传入配置化的路由映射Map做匹配
                if ("type1".equals(routeValue)) {
                    targetTopic = "topic1";
                } else if ("type2".equals(routeValue)) {
                    targetTopic = "topic2";
                }
            }
        } catch (Exception e) {
            // 解析异常时走兜底Topic，可按需添加错误计数、日志打印逻辑
            targetTopic = defaultTopic;
        }
        // 复用原有的SimpleStringSchema做value序列化，保证UTF-8编码一致
        byte[] value = valueSerializer.serialize(element);
        // 如需指定Kafka消息key，可在此处补充key的序列化逻辑，传入ProducerRecord的key参数
        return new ProducerRecord<>(targetTopic, value);
    }
}

替换原有代码中的KafkaSink构造部分即可：

DataStream<String> data = .....
KafkaSink<String> sink = KafkaSink.<String>builder()
        .setBootstrapServers(parameter.get("bootstrap.servers"))
        // 替换原来的固定Topic序列化器为自定义动态路由实现
        .setRecordSerializer(new DynamicKafkaTopicSerializer(
                parameter.get("kafka.output.default.topic"),
                "your_route_key" // 替换为JSON中用来判断路由的实际字段名
        ))
        .setDeliverGuarantee(DeliveryGuarantee.AT_LEAST_ONCE)
        .build();
data.sinkTo(sink);

注意事项 #

• 不要在serialize方法中执行外部IO（比如查数据库、远程接口拿路由规则），该方法会按消息条数高频调用，外部IO会成为吞吐瓶颈。如果需要动态更新路由规则，可以将规则通过广播流下发到算子本地缓存，直接在内存中匹配即可。
• 如果JSON结构复杂、吞吐要求高，可以替换为Jackson的流式解析API，只读取路由字段的值，不用解析整个JSON结构，能进一步提升性能。
• 该实现完全兼容KafkaSink的所有投递语义（至少一次、精确一次），不需要调整其他sink配置。
• 如果路由规则需要配置化，可在构造自定义序列化器时传入路由映射Map，不用在代码里硬编码匹配逻辑。

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