配置 Debezium CDC -> Kafka -> JDBC Sink (Multiple tables) 问题答案

【问题标题】：Configure Debezium CDC -> Kafka -> JDBC Sink (Multiple tables) Question配置 Debezium CDC -> Kafka -> JDBC Sink (Multiple tables) 问题
【发布时间】：2023-01-06 23:07:27
【问题描述】：

我们在 SQL Server DB（应用程序数据库）中有大约 100 个表，需要近乎实时地同步到 SQL Server DB（用于分析）。

未来用例：近乎实时地将 30 个源数据库的概念证明扩展到一个目标数据库（用于分析）。

我正在考虑为多个表使用一个接收器连接器或几个接收器连接器。如果这是个好主意，请告诉我。

但我不确定如何配置接收器以适应多个表，尤其是每个表可能都有自己的主键。互联网似乎有非常简单的接收器连接器示例，但没有解决复杂的用例。

Debezium CDC（源代码）配置

  {  "name": "wwi",
    "config": {
     "connector.class": "io.debezium.connector.sqlserver.SqlServerConnector",
  "database.dbname": "************************",
  "database.history": "io.debezium.relational.history.MemoryDatabaseHistory",
  "database.hostname": "**********************",
  "database.password": "**********************",
  "database.port": "1433",
  "database.server.name": "******",
  "database.user": "*********",
  "decimal.handling.mode": "string",
  "key.converter": "org.apache.kafka.connect.json.JsonConverter",
  "key.converter.schemas.enable": "true",
  "snapshot.mode": "schema_only",
  "table.include.list": "Sales.Orders,Warehouse.StockItems",
  "tasks.max": "1",
  "tombstones.on.delete": "false",
  "transforms": "route,unwrap",
  "transforms.route.regex": "([^.]+)\\.([^.]+)\\.([^.]+)",
  "transforms.route.replacement": "$3",
  "transforms.route.type": "org.apache.kafka.connect.transforms.RegexRouter",
  "transforms.unwrap.type": "io.debezium.transforms.ExtractNewRecordState",
  "value.converter.schemas.enable": "true",
  "value.convertor": "org.apache.kafka.connect.json.JsonConverter"
    }
    }

JDBC 接收器配置

{
    "name": "sqlsinkcon",
    "config": {
        "connector.class": "io.confluent.connect.jdbc.JdbcSinkConnector",
        "topics": "orders",
        "tasks.max": "1",
        "auto.evolve": "true",
        "connection.user": "********",
        "auto.create": "true",
        "connection.url": "jdbc:sqlserver://************",
        "insert.mode": "upsert",
        "pk.mode":"record_key",
        "pk.fields":"OrderID",
        "db.name": "kafkadestination"
    }
}

【问题讨论】：

标签： apache-kafka apache-kafka-connect debezium

【解决方案1】：

接收器将为每个消耗的主题写入一个表。 topics 或 topics.regex 可用于一次消费多个主题。

关于可伸缩性（或至少是容错性），我更喜欢一个接收器任务，一个主题（因此写入一个表）。否则，如果您使用多个主题，并且连接器失败，那么由于消费者重新平衡，它可能会导致所有任务线程崩溃。

此外，就网络带宽而言，在 Kafka 中使用 JSON / 纯文本格式并不是最佳选择。我建议使用像 Avro 或 Protobuf 这样的二进制格式。

【讨论】：