我有一个mysql表,如下所示:
我使用kafka连接器将此表添加到kafka主题:
ksql> CREATE SOURCE CONNECTOR SOURCE_MYSQL_01 WITH (
'connector.class' = 'io.debezium.connector.mysql.MySqlConnector',
'database.hostname' = 'mysql',
'database.port' = '3306',
'database.user' = 'debezium',
'database.password' = 'dbz',
'database.server.id' = '42',
'database.server.name' = 'asgard',
'table.whitelist' = 'demo.customers',
'database.history.kafka.bootstrap.servers' = 'kafka:29092',
'database.history.kafka.topic' = 'dbhistory.demo' ,
'include.schema.changes' = 'false',
'transforms'= 'unwrap,extractkey',
'transforms.unwrap.type'= 'io.debezium.transforms.ExtractNewRecordState',
'transforms.extractkey.type'= 'org.apache.kafka.connect.transforms.ExtractField$Key',
'transforms.extractkey.field'= 'id',
'key.converter'= 'org.apache.kafka.connect.storage.StringConverter',
'value.converter'= 'io.confluent.connect.avro.AvroConverter',
'value.converter.schema.registry.url'= 'http://schema-registry:8081'
);
然后我想用它来创建一个基于它的表:
CREATE TABLE CUSTOMERS WITH (KAFKA_TOPIC='asgard.demo.CUSTOMERS', VALUE_FORMAT='AVRO');
然后我得到这个错误:
Tables require a PRIMARY KEY. Please define the PRIMARY KEY.
Use a partial schema to define the primary key and still load the value columns from the Schema Registry, for example:
CREATE TABLE CUSTOMERS (ID INT PRIMARY KEY) WITH (...);
当我按照建议更改时:
CREATE TABLE CUSTOMERS (id int primary key) WITH (KAFKA_TOPIC='asgard.demo.CUSTOMERS', VALUE_FORMAT='AVRO');
我得到这个错误:
Duplicate column names: `ID`
做了些搜索,但还是卡在这里。这有什么问题,我怎么解决?谢谢
1条答案
按热度按时间kxe2p93d1#
问题是,您分配给pk列的名称与从模式注册表加载的avro模式中的列名称冲突。
您可以随意命名键列,因为列名在任何地方都不会持久化,所以只需将其命名为不冲突的名称即可。
customer_id
.另请注意:
'key.converter'='org.apache.kafka.connect.storage.stringconverter'
我相信这会把钥匙变成
STRING
. 不确定这是故意的。ksqldb将与INT
pk太。。。