为数据库中的多个表配置debezium连接器

Question

我正在尝试为MySQL数据库中的多个表配置Debezium连接器（我在MySQL 8.0上使用Debezium 1.4）。在Kafka中创建主题时，我的公司有一个命名模式要遵循，而这个模式不允许使用下划线（u），所以我不得不用连字符（-）替换它们

所以，我的主题名称是：

专题1

fjf.db.top-domain.domain.sub-domain.transaction-search.order-status
WHERE
- transaction-search = schema "transaction_search"
- order-status = table "order_status". 
- All changes in that table, must go to that topic.

专题二

fjf.db.top-domain.domain.sub-domain.transaction-search.shipping-tracking
WHERE
- transaction-search = schema "transaction_search"
- shipping-tracking = table "shipping_tracking"
- All changes in that table, must go to that topic.

专题三

fjf.db.top-domain.domain.sub-domain.transaction-search.proposal
WHERE
- transaction-search = schema "transaction_search"
- proposal = table "proposal"
- All changes in that table, must go to that topic.

我正在尝试使用转换“ByLogicalTableRouter”，但我找不到解决我的问题的正则表达式解决方案。

{ "name": "debezium.connector",
 "config":
    { 
"connector.class": "io.debezium.connector.mysql.MySqlConnector",
"tasks.max": "1",
"database.hostname": "myhostname",
"database.port": "3306",
"database.user": "debezium", 
"database.password": "password", 
"database.server.id": "1000", 
"database.server.name": "fjf.db.top-domain.domain.sub-domain.transaction-search",
"schema.include.list": "transaction_search",
"table.include.list": "transaction_search.order_status,transaction_search.shipping_tracking,transaction_search.proposal",
"database.history.kafka.bootstrap.servers": "kafka.intranet:9097",
"database.history.kafka.topic": "fjf.db.top-domain.domain.sub-domain.transaction-search.schema-history",
"snapshot.mode": "schema_only",
"transforms":"RerouteName,RerouteUnderscore",
"transforms.RerouteName.type":"io.debezium.transforms.ByLogicalTableRouter",
"transforms.RerouteName.topic.regex":"(.*)transaction_search(.*)",
"transforms.RerouteName.topic.replacement": "$1$2" 
"transforms.RerouteUnderscore.type":"io.debezium.transforms.ByLogicalTableRouter",
"transforms.RerouteUnderscore.topic.regex":"(.*)_(.*)",
"transforms.RerouteUnderscore.topic.replacement": "$1-$2" 
    }
}

• 在第一个转换中，我试图删除主题路由中重复的模式名称。
• 在第二个转换中，替换所有的残余物下划线_hiphens-

但有了这些，我得到了下面的错误，这表明它试图将所有内容发送到同一个主题

原因：org。阿帕奇。Kafka。连接错误。SchemaBuilderException:无法创建字段，因为字段名重复_dbz__physicalTableIdentifier

如何进行转换，将每个表的事件转发到各自的主题？

Answer 1

1. 删除架构名称

在第一个转换中，我试图删除主题路由中重复的模式名称。

使用正则表达式转换后，将有两个点，因此需要修复：

"transforms.RerouteName.topic.regex":"([^.]+)\\.transaction_search\\.([^.]+)",
"transforms.RerouteName.topic.replacement": "$1.$2" 

您可以尝试使用来自Kafka Connect Common Transformation的ChangeCase SMT。