如何通过Debezium配置按表字段值将消息发送到Kafka指定分区

当然可以实现这个需求，以下是两种可行的方案：

方案一：自定义Kafka分区器（精准映射字段到指定分区） #

这是实现严格字段值到固定分区映射的最佳方式，核心是编写自定义分区逻辑，让Debezium的Kafka生产者使用这个逻辑分配分区：

• 编写一个实现org.apache.kafka.clients.producer.Partitioner接口的类，在partition方法中解析消息里的customer字段，根据字段值返回对应的分区编号。
• 在Debezium连接器配置中添加参数，指定使用这个自定义分区器：

  producer.partitioner.class=com.yourpackage.CustomerPartitioner
  

• 自定义分区器的核心代码示例：

  public class CustomerPartitioner implements Partitioner {
      @Override
      public int partition(String topic, Object key, byte[] keyBytes, Object value, byte[] valueBytes, Cluster cluster) {
          // 从Debezium的ChangeEvent中提取customer字段值
          Struct structValue = (Struct) value;
          String customer = structValue.getString("customer");
          
          // 按需求映射到指定分区
          switch(customer) {
              case "customer1": return 0;
              case "customer2": return 1;
              case "customer3": return 2;
              case "customer4": return 3;
              // 处理未匹配的情况，默认按键哈希分配
              default: return Math.abs(key.hashCode()) % cluster.partitionCountForTopic(topic);
          }
      }
  
      @Override
      public void close() {}
  
      @Override
      public void configure(Map<String, ?> configs) {}
  }
  

• 注意：要确保自定义分区器的类文件能被Debezium的类加载器访问到，比如将其打包到Debezium的插件目录，或者构建包含该类的自定义Debezium镜像。

方案二：提取字段作为消息键（同字段值消息进入同一分区） #

如果不需要严格固定分区编号，只要求相同customer值的消息进入同一个分区，可以利用Kafka默认的分区策略，将customer字段设置为消息的键：

• 在Debezium连接器配置中添加以下参数，通过Kafka Connect的Transform功能提取customer字段作为消息键：

  # 配置键的序列化器
  key.converter=org.apache.kafka.connect.json.JsonConverter
  key.converter.schemas.enable=false
  
  # 配置Transform提取customer字段为键
  transforms=ExtractCustomerKey
  transforms.ExtractCustomerKey.type=org.apache.kafka.connect.transforms.ExtractField$Key
  transforms.ExtractCustomerKey.field=customer
  

• 原理：Kafka默认分区器会根据消息键的哈希值计算分区，相同customer值的键会生成相同的哈希，从而被分配到同一个分区（前提是Kafka主题的分区数足够，且哈希分布均匀）。

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