如何根据Kafka Connect消息值动态切换PostgreSQL数据库Schema？

方案一：利用PostgreSQL Sink Connector动态表达式+内置SMT #

Confluent官方的PostgreSQL Sink Connector（版本5.5+）支持EL表达式动态配置目标Schema，结合内置的Single Message Transform（SMT）可直接实现需求：

1. 核心配置步骤

   • 开启连接器的表达式支持：设置enable.expressions=true
   • 动态指定Schema名称：将schema.name配置为country_${value.cityId}，其中${value.cityId}会自动从每条消息的value字段提取cityId值，拼接成对应Schema（如country_1、country_2）
   • 确保消息字段可被解析：若使用JSON消息，配置value.converter=org.apache.kafka.connect.json.JsonConverter，并设置value.converter.schemas.enable=false（如果消息不带Schema信息）

2. 示例配置片段

   name=postgres-city-sink
   connector.class=io.confluent.connect.jdbc.JdbcSinkConnector
   tasks.max=1
   topics=country_city
   connection.url=jdbc:postgresql://localhost:5432/your-db
   connection.user=db-user
   connection.password=db-pass
   auto.create=false
   auto.evolve=false
   enable.expressions=true
   schema.name=country_${value.cityId}
   table.name.format=city
   value.converter=org.apache.kafka.connect.json.JsonConverter
   value.converter.schemas.enable=false
   

方案二：自定义Single Message Transform（SMT） #

如果cityId和Schema名称不是简单的拼接关系，可自定义SMT实现复杂映射逻辑：

1. 自定义SMT核心逻辑

   • 实现org.apache.kafka.connect.transforms.Transformation接口
   • 在apply方法中提取消息的cityId字段
   • 根据自定义映射规则（如配置文件中定义的映射关系）生成目标Schema名称
   • 通过添加自定义头部字段，让PostgreSQL Sink读取该头部确定Schema

2. 核心代码片段

   public class DynamicSchemaTransform<R extends ConnectRecord<R>> implements Transformation<R> {
       private Map<String, String> schemaMapping;
   
       @Override
       public R apply(R record) {
           Struct value = (Struct) record.value();
           String cityId = value.getString("cityId");
           String targetSchema = schemaMapping.getOrDefault(cityId, "default_schema");
           // 添加自定义头部供连接器读取
           return record.newRecord(
               record.topic(),
               record.kafkaPartition(),
               record.keySchema(),
               record.key(),
               record.valueSchema(),
               record.value(),
               record.timestamp(),
               record.headers().addString("target.schema", targetSchema)
           );
       }
   
       @Override
       public void configure(Map<String, ?> configs) {
           // 从配置加载cityId到Schema的映射
           schemaMapping = (Map<String, String>) configs.get("schema.mapping");
       }
   
       // 其余接口方法实现省略
   }
   

3. 连接器配合配置
   将自定义SMT放入Kafka Connect插件目录，然后添加以下配置：

   transforms=dynamicSchema
   transforms.dynamicSchema.type=com.yourcompany.DynamicSchemaTransform
   transforms.dynamicSchema.schema.mapping=1:country_1,2:country_2,3:country_3
   schema.name=${kafka.headers.target.schema}
   enable.expressions=true
   

方案三：Kafka Streams前置处理消息 #

若需要复杂业务逻辑（如多字段组合确定Schema、数据清洗），可先用Kafka Streams预处理消息：

1. Streams处理逻辑

   • 消费country_city主题消息
   • 提取cityId生成目标Schema名称，将其作为新字段（如targetSchema）写入消息
   • 将处理后的消息发送到原主题或中间主题，供Kafka Connect消费

2. 示例Streams代码片段

   KStream<String, JsonNode> source = builder.stream("country_city");
   source.mapValues(value -> {
       int cityId = value.get("cityId").asInt();
       ObjectNode newNode = ((ObjectNode) value);
       newNode.put("targetSchema", "country_" + cityId);
       return newNode;
   }).to("country_city_processed");
   

3. Kafka Connect配置
   消费处理后的主题，配置schema.name=${value.targetSchema}并开启enable.expressions=true即可实现动态切换。

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