谢谢回答 使用新属性可以 成功修改记录 , 但是不太明白 "使用老的option参数还是需要根据query推导主键" 这里话是什么意思,需要怎么做
在 2020-07-31 16:46:41，"Leonard Xu" <xbjt...@gmail.com> 写道： >Hi, chenxuying > >看你还是用的还是 " 'connector.type' = 'jdbc', …. " >，这是老的option，使用老的option参数还是需要根据query推导主键， >需要使用新的属性[1]：" 'connector' = 'jdbc’,…." 才能配合 主键 决定 upsert 模式. > >Best >Leonard >[1] >https://ci.apache.org/projects/flink/flink-docs-release-1.11/zh/dev/table/connectors/jdbc.html#connector-options > ><https://ci.apache.org/projects/flink/flink-docs-release-1.11/zh/dev/table/connectors/jdbc.html#connector-options> > >> 在 2020年7月31日，16:12，chenxuying <cxydeve...@163.com> 写道： >> >> hi >> 我使用的flink 1.11.0版本 >> 代码如下 >> StreamExecutionEnvironment streamEnv = >> StreamExecutionEnvironment.getExecutionEnvironment(); >> TableEnvironment tableEnvironment = StreamTableEnvironment.create(streamEnv); >> tableEnvironment.executeSql(" " + >> " CREATE TABLE mySource ( " + >> " a bigint, " + >> " b bigint " + >> " ) WITH ( " + >> " 'connector.type' = 'kafka', " + >> " 'connector.version' = 'universal', " + >> " 'connector.topic' = 'mytesttopic', " + >> " 'connector.properties.zookeeper.connect' = '172.17.0.2:2181', " + >> " 'connector.properties.bootstrap.servers' = '172.17.0.2:9092', " + >> " 'connector.properties.group.id' = 'flink-test-cxy', " + >> " 'connector.startup-mode' = 'latest-offset', " + >> " 'format.type' = 'json' " + >> " ) "); >> tableEnvironment.executeSql("CREATE TABLE mysqlsink ( " + >> " id bigint, " + >> " game_id varchar, " + >> " PRIMARY KEY (id) NOT ENFORCED " + >> " ) " + >> " with ( " + >> " 'connector.type' = 'jdbc', " + >> " 'connector.url' = 'jdbc:mysql://47.99.181.86:3306/flinksql?useSSL=false' >> , " + >> " 'connector.username' = 'root' , " + >> " 'connector.password' = 'root', " + >> " 'connector.table' = 'mysqlsink' , " + >> " 'connector.driver' = 'com.mysql.cj.jdbc.Driver' , " + >> " 'connector.write.flush.interval' = '2s', " + >> " 'connector.write.flush.max-rows' = '300' " + >> " )"); >> tableEnvironment.executeSql("insert into mysqlsink (`id`,`game_id`) values >> (select a,cast(b as varchar) b from mySource)"); >> >> >> 问题一 : 上面的insert语句会出现如下错误 >> Caused by: org.apache.calcite.sql.validate.SqlValidatorException: Cannot >> apply '$SCALAR_QUERY' to arguments of type '$SCALAR_QUERY(<RECORDTYPE(BIGINT >> A, VARCHAR(2147483647) B)>)'. Supported form(s): >> '$SCALAR_QUERY(<RECORDTYPE(SINGLE FIELD)>)' >> >> >> 问题二 : 如果insert改成 tableEnvironment.executeSql("insert into mysqlsink select >> a,cast(b as varchar) b from mySource"); 是可以运行的,但是出现唯一主键重复时会报错 >> Caused by: java.sql.SQLIntegrityConstraintViolationException: Duplicate >> entry '1' for key 'PRIMARY' >> >> >> >
