设置:我在mssqlserver上启用了cdc,cdc事件通过debezium-kafka-connect(源代码)反馈给kafka。在kafka中,多个表cdc事件被路由到单个主题。
问:因为kafka主题中有多个表数据,所以我希望在cdc数据中有表名和数据库名。
我在mysql cdc中得到表名和数据库名,但在mssqlcdc中没有。
下面是sql server的debezium源连接器
curl -i -X POST -H "Accept:application/json" -H "Content-Type:application/json" localhost:8083/connectors/ -d '{
"name": "cdc-user_profile-connector",
"config": {
"connector.class": "io.debezium.connector.sqlserver.SqlServerConnector",
"tasks.max": "1",
"database.hostname": "<<hostname>>",
"database.port": "<<port>>",
"database.user": "<<username>>",
"database.password": "<<password>>",
"database.server.name": "test",
"database.dbname": "testDb",
"table.whitelist": "schema01.table1,schema01.table2",
"database.history.kafka.bootstrap.servers": "broker:9092",
"database.history.kafka.topic": "digital.user_profile.schema.audit",
"database.history.store.only.monitored.tables.ddl": true,
"include.schema.changes": false,
"event.deserialization.failure.handling.mode": "fail",
"snapshot.mode": "initial_schema_only",
"snapshot.locking.mode": "none",
"transforms":"addStaticField,topicRoute",
"transforms.addStaticField.type":"org.apache.kafka.connect.transforms.InsertField$Value",
"transforms.addStaticField.static.field":"source_system",
"transforms.addStaticField.static.value":"source_system_1",
"transforms.topicRoute.type":"org.apache.kafka.connect.transforms.RegexRouter",
"transforms.topicRoute.regex":"(.*)",
"transforms.topicRoute.replacement":"digital.user_profile",
"errors.tolerance": "none",
"errors.log.enable": true,
"errors.log.include.messages": true,
"errors.retry.delay.max.ms": 60000,
"errors.retry.timeout": 300000
}
}'
我得到以下输出(演示数据)
{
"before": {
"profile_id": 147,
"email_address": "test@gmail.com"
},
"after": {
"profile_id": 147,
"email_address": "test_modified@gmail.com"
},
"source": {
"version": "0.9.4.Final",
"connector": "sqlserver",
"name": "test",
"ts_ms": 1556723528917,
"change_lsn": "0007cbe5:0000b98c:0002",
"commit_lsn": "0007cbe5:0000b98c:0003",
"snapshot": false
},
"op": "u",
"ts_ms": 1556748731417,
"source_system": "source_system_1"
}
我的要求如下
{
"before": {
"profile_id": 147,
"email_address": "test@gmail.com"
},
"after": {
"profile_id": 147,
"email_address": "test_modified@gmail.com"
},
"source": {
"version": "0.9.4.Final",
"connector": "sqlserver",
"name": "test",
"ts_ms": 1556723528917,
"change_lsn": "0007cbe5:0000b98c:0002",
"commit_lsn": "0007cbe5:0000b98c:0003",
"snapshot": false,
"db": "testDb",
"table": "table1/table2"
},
"op": "u",
"ts_ms": 1556748731417,
"source_system": "source_system_1"
}
2条答案
按热度按时间kmpatx3s1#
这是计划的一部分https://issues.jboss.org/browse/dbz-875 问题
pokxtpni2#
debezium kafka connect通常将每个表中的数据放在单独的主题中,主题名称的格式为hostname.database.table。我们通常使用主题名来区分源表和数据库名。
如果要将所有表中的数据手动放到一个主题中,那么可能还需要手动添加表和数据库名称。