#遇到的问题
flinksql的通过flinkcdc从mysql读取一张表,写入kafka,报错信息:
Exception in thread "main" org.apache.flink.table.api.TableException: Table sink 'default_catalog.default_database.kafka_t4' doesn't support consuming update and delete changes which is produced by node TableSourceScan(table=[[default_catalog, default_database, cdc_t4]], fields=[id, name])
#运行的代码
tableEnv.executeSql(
"CREATE TABLE cdc_t4 (\n" +
" id bigint,\n" +
" name varchar,\n" +
" PRIMARY KEY (id) NOT ENFORCED\n" +
" ) WITH (\n" +
" 'connector' = 'mysql-cdc',\n" +
" 'hostname' = 'cnap421',\n" +
" 'port' = '3306',\n" +
" 'server-time-zone' = 'Asia/Shanghai',\n" +
" 'debezium.snapshot.mode' = 'initial',\n" + //initial,earliest,latest
" 'username' = 'root',\n" +
" 'password' = 'sgs123654.',\n" +
" 'database-name' = 'test',\n" +
" 'table-name' = 't4'\n" +
" )"
);
tableEnv.executeSql(
"create table kafka_t4(id bigint, name varchar"
+ ") with(\n"
+ "'connector' = 'kafka',\n"
+ "'topic' = 'test_output_sink',\n"
+ "'properties.bootstrap.servers' = '10.168.136.31:9092,10.168.136.28:9092',\n"
+ "'format' = 'json'\n"
+ ")"
);
tableEnv.executeSql(
"insert into kafka_t4 select * from cdc_t4"
);
#解决思路
将connect的kafka换成upsert-kafka就可以了,我以前一直使用flink的datastream的api方式处理的,没注意这个。
#现在的疑问
flinksql的方式通过flinkcdc将mysql的binlog写入kafka,一定要用upsert-kafka吗