如何通过Kafka无需为每张表创建连接器，实现MySQL数据同步至ES及其他数据库？

当然有办法解决这个重复建连接器的麻烦！相信很多做CDC（变更数据捕获）同步的同学都遇到过这种痛点，下面给你几个实用的实现思路：

1. 单源连接器捕获多表变更 #

MySQL Source Connector本身就支持通过通配符或正则表达式批量指定要同步的表，完全不用逐个配置：

• 如果你要同步某个数据库下的所有表，只需要在连接器配置里这么写：

  database.include.list = your_target_db
  table.include.list = your_target_db.*
  

• 如果只需要同步符合特定规则的表（比如前缀为order_的业务表），可以用正则匹配：

  table.include.list = your_target_db.order_.*
  

这样一个源连接器就能自动捕获所有匹配表的增删改变更，不用重复创建多个源连接器。

2. 单Sink连接器处理多表数据 #

搞定源端之后，Sink端也可以用单个连接器处理所有表的同步，不同目标系统的配置思路略有不同：

针对Elasticsearch Sink #

Elasticsearch Sink Connector支持动态索引映射，可以自动根据源表名生成对应的ES索引，配置示例：

topics.regex = your_target_db\\..*
key.converter = org.apache.kafka.connect.storage.StringConverter
value.converter = io.confluent.connect.avro.AvroConverter
value.converter.schema.registry.url = http://schema-registry:8081
connection.url = http://elasticsearch:9200
type.name = _doc
index.name.format = ${topic.replace("your_target_db.", "")}

这里topics.regex匹配源连接器输出的所有主题（默认格式是数据库名.表名），index.name.format会自动把主题里的数据库前缀去掉，用表名作为ES索引名，实现一张表对应一个索引，全程只用一个Sink连接器。

针对其他关系型数据库Sink（比如PostgreSQL） #

如果目标是另一个关系型数据库，可以用动态表名路由。以JDBC Sink Connector为例：

topics.regex = your_target_db\\..*
connection.url = jdbc:postgresql://postgres:5432/backup_db
connection.user = db_user
connection.password = db_pass
auto.create = true
auto.evolve = true
table.name.format = ${topic.replace("your_target_db.", "")}

同样通过topics.regex匹配所有主题，table.name.format提取出表名后，会自动在目标库创建对应表并同步数据，一个Sink就能搞定全量表的备份。

3. 额外优化小技巧 #

• 如果需要对特定表做个性化处理（比如字段映射、数据过滤），可以在Sink连接器里配置transforms，用正则匹配特定表的主题来单独应用转换规则。
• 可以结合Kafka的主题分区策略，让每张表的数据对应单独的分区，既提升同步效率，也方便后续排查问题。

这样一套配置下来，只需要1个源连接器+1个Sink连接器，就能实现整个数据库（或指定规则表）的变更同步，既简化了操作，也能保证备份和查询服务的响应速度。

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