在 Debezium 中无法根据 MySQL 的表创建某些主题

Posted 2023-03-23

【中文标题】在 Debezium 中无法根据 MySQL 的表创建某些主题

【英文标题】：Some topics are unable to created as per tables of MySQL in Debezium

【发布时间】：2021-12-15 00:20:23

【问题描述】：

我使用 Debezium for CDC 从 mysql 到 Kafka 主题，这里的问题是创建了一些主题，而一些主题没有根据特定数据库中的表列表创建。 我在数据库中有 5 个表，名称是 demo。

table_1、table_2、table_3、table_4 和 table_5。

能够创建 2 个表格，例如：

demo.demo.table_1
demo.demo.table_2

预期的主题应该是：

 demo.demo.table_1
 demo.demo.table_2
 demo.demo.table_3
 demo.demo.table_4
 demo.demo.table_5

根据 Debezium 的文档：

默认行为是 Debezium MySQL 连接器将一张表中所有 INSERT、UPDATE 和 DELETE 操作的事件写入一个 Kafka 主题。 Kafka主题命名约定如下：

serverName.databaseName.tableName

假设fulfillment 是服务器名称，inventory 是数据库名称，并且数据库包含名为orders、customers 和products 的表。 Debezium MySQL 连接器向三个 Kafka 主题发出事件，每个主题对应一个数据库中的表：

fulfillment.inventory.orders
fulfillment.inventory.customers
fulfillment.inventory.products

无法找到问题到底发生在哪里，我应该在哪里检查问题？ 任何日志或权限问题。 此用户拥有所有权限。

下面是配置.JSON文件。

  "name": "inventory-connector",  
  "config":   
    "connector.class": "io.debezium.connector.mysql.MySqlConnector",
    "tasks.max": "1",  
    "database.hostname": "localhost",  
    "database.port": "3306",
    "database.user": "agent",
    "database.password": "123",
    "database.server.id": "123",  
    "database.allowPublicKeyRetrieval":"true",
    "database.server.name": "demo",  
    "database.include.list": "demo",  
    "database.history.kafka.bootstrap.servers": "localhost:9092",  
    "database.history.kafka.topic": "schema-changes.demo"  
  

如果有人有任何解决方案，请建议我。

谢谢， 双马尔。

【参考方案1】：

  "name": "inventory-connector",  
  "config":   
    "connector.class": "io.debezium.connector.mysql.MySqlConnector",
    "tasks.max": "1",  
    "database.hostname": "localhost",  
    "database.port": "3306",
    "database.user": "agent",
    "database.password": "123",
    "database.server.id": "123",  
    "database.allowPublicKeyRetrieval":"true",
    "database.server.name": "demo",  
    "database.include.list": "demo",  
    "database.history.kafka.bootstrap.servers": "localhost:9092",  
    "database.history.kafka.topic": "schema-changes.demo",
    "name": "inventory-connector",
    "table.include.list": "table_1, table_2, table_3, table_4, table_5",
    "snapshot.mode": "schema_only"
  

我想你错过了 "table.include.list" debezium 默认 "snapshot.mode" 为 initial，它将从 mysql binlog 中数据的开头获取所有快照。如果 mysql binlog 大小为 1 GB，那么您必须耐心查看结果，kafka broker 仍在处理日志...

我建议你使用"snapshot.mode": "schema_only"，它只会对schema进行快照，并且只会记录连接器启动时的数据。

并尝试更新每个表中的数据，主题将自动生成

我希望它有效

问候

【讨论】：

感谢您的回复。在运行这两个命令时面临一个问题，1：“ bin/connect-distributed.sh config/connect-distributed.properties ”和 2：“ curl -i -X POST -H “Accept: application/json” -H “ Content-Type: application/json" localhost:8083/connectors -d @src-srink.json " 第一个命令不起作用，它没有给出任何控制台消息，表示错误或类似信息的简单移动下一行。