Debezium CDC-Kafka-JDBC Sink多表同步配置咨询

一、单个/少量JDBC Sink连接器处理多表同步的可行性 #

完全可行，但需结合负载、扩展性做合理配置：

• 当前100张表场景：单Sink配合多任务（tasks.max）即可支撑，只要Kafka Connect集群资源足够，可通过并行任务分摊各表的同步压力。
• 未来30个源库扩展场景：需确保每个Debezium源连接器配置独立的database.server.name（避免不同源库的topic命名冲突），Sink端通过正则匹配所有需同步的topic，少量Sink即可覆盖需求，但要根据总表数量和数据量调整任务数与集群资源。

需注意的风险点：

• 若单Sink任务数过高，可能引发数据库连接池压力过大，需同步调整JDBC连接池参数（如connection.max.connections）。
• 高并发场景下，需监控各topic的同步延迟，避免某张热点表拖慢整体同步效率。

二、多表同步（主键独立）的Sink配置调整 #

针对各表主键独立的需求，核心是让Sink自动识别每张表的主键，无需手动指定固定主键字段。结合你现有配置，调整如下：

1. 源端配置修正（先解决现有问题） #

你的Debezium配置中存在拼写错误："value.convertor"应改为"value.converter"，否则会导致序列化失败。

2. 调整后的JDBC Sink配置示例 #

{
    "name": "sqlsinkcon",
    "config": {
        "connector.class": "io.confluent.connect.jdbc.JdbcSinkConnector",
        "topics.regex": ".*", // 匹配所有需同步的表topic，若需指定表可改为"Orders,StockItems,Table3,..."
        "tasks.max": "5", // 根据表数量与负载调整，建议不超过同步表总数的1/2
        "auto.evolve": "true",
        "connection.user": "********",
        "auto.create": "true",
        "connection.url": "jdbc:sqlserver://************",
        "insert.mode": "upsert",
        "pk.mode": "record_key", // 自动从Kafka消息的record key提取主键，适配各表独立主键
        "table.name.format": "${topic}", // 目标表名与topic名一致，若需指定schema可改为"Sales.${topic}"
        "db.name": "kafkadestination",
        "connection.max.connections": "10" // 根据任务数调整数据库连接池大小
    }
}

3. 关键配置说明 #

• topics.regex/topics：用正则匹配所有需同步的表对应的topic（你源端通过RegexRouter将topic转为表名，所以直接匹配所有表名即可）；若需精确指定表，用逗号分隔表名。
• pk.mode: record_key：Debezium捕获CDC时会自动将源表的主键写入Kafka消息的record key中，Sink会自动识别该key作为目标表的主键，无需手动指定pk.fields，完美适配各表主键独立的场景。
• table.name.format：控制目标表的命名规则，若源端topic是表名，目标库表名与源表一致则用${topic}；若目标库需要加schema前缀，可改为${schema}.${topic}（需结合源端路由调整）。
• tasks.max：设置并行任务数，让Kafka Connect同时处理多个表的同步，提升整体效率。

4. 源端扩展适配（未来30个源库） #

每个源库的Debezium连接器需设置唯一的database.server.name，例如：

"database.server.name": "source_db_01"

此时源端生成的topic会是source_db_01.Sales.Orders，若需统一topic命名方便Sink过滤，可修改RegexRouter的规则：

"transforms.route.regex": "([^.]+)\\.([^.]+)\\.([^.]+)",
"transforms.route.replacement": "$1_$3" // 将topic转为source_db_01_Orders

Sink端则用topics.regex: "source_db_.*"匹配所有源库的表topic。

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