使用flink sql 将kafka的数据同步到mysql无法删除。

Thu, 23 Feb 2023 22:20:15 -0800 

-----建表语法如下
String kafka = "CREATE TABLE `电话` " +
"(    `rowID` VARCHAR(255),`名称` STRING,`手机` VARCHAR(255),`座机` VARCHAR(255),  " +
"  PRIMARY KEY (`rowID`) NOT ENFORCED  ) " +
" WITH " +
"(    'connector' = 'jdbc',   " +
" 'driver' = 'com.mysql.cj.jdbc.Driver',   " +
" 'url' = 'jdbc:mysql://XXXXXX:6506/meihua_test',  " +
"  'username' = 'root',  " +
"  'password' = '123456',  " +
"  'table-name' = '电话'  )";

String mysql = "CREATE TABLE `电话_1` " +
"(    `rowid` VARCHAR(100)," +
"`63f73b332e77497da91286f0` VARCHAR(100)," +
"`63f73b3f2e77497da91286fb` VARCHAR(100)," +
"`63f73b3f2e77497da91286fc` VARCHAR(100)," +
"`op` STRING ," +
" PRIMARY KEY (rowid) NOT ENFORCED )" +
" WITH " +
"( 'connector' = 'kafka', " +
"'topic' = 'sz_worksheet-63f82984f3ec743e45b0d561-63f73b332e77497da91286ef'," +
" 'properties.bootstrap.servers' = 'XXXXXX:9092'," +
" 'scan.startup.mode' = 'earliest-offset', " +
"'format' = 'debezium-json' )";
-----执行语句如下
String insert = "insert into `电话` select `t_1`.`rowID` as `rowID`,`t_1`.`名称` as 
`名称`,`t_1`.`手机` as `手机`,`t_1`.`座机` as `座机` from" +
" ( select `rowid` as `rowID`,`63f73b332e77497da91286f0` as 
`名称`,`63f73b3f2e77497da91286fb` as `手机`,`63f73b3f2e77497da91286fc` as `座机` from 
`电话_1` ) as t_1";
-----操作数据如下


String insert = "insert into `电话` select `t_1`.`rowID` as `rowID`,`t_1`.`名称` as 
`名称`,`t_1`.`手机` as `手机`,`t_1`.`座机` as `座机` from" +
" ( select `rowid` as `rowID`,`63f73b332e77497da91286f0` as 
`名称`,`63f73b3f2e77497da91286fb` as `手机`,`63f73b3f2e77497da91286fc` as `座机` from 
`电话_1` ) as t_1";
-----执行语句如下
{
    "op":"d",
    "before":{
        "rowid":"f251af39-1a95-4d6f-b4cb-cdf93d5d1b6d"
    },
    "after":null
}
现在的结论是可以新增和修改，但是无法删除。难道insert into这个语句搞不定吗？ 走的debezuim json序列化的格式。
各位大佬帮看下 谢谢。

回复