?????? ????flink-jdbc-driver????mysql??????(flink1.10.0)

Tue, 25 Feb 2020 02:21:47 -0800 

??????????????????


????????????????????????mysql????????????UpsertStreamTableSink????
????????????????????????????????????????????????????????????????


??????????????????????????????????????????insert into t_user_target 
values(1,'fan'),(2,'ss')??sink??
????????????????????????sink??????????sql??????????????????????sink??


------------------ ???????? ------------------
??????:&nbsp;"Jark Wu"<[email protected]&gt;;
????????:&nbsp;2020??2??25??(??????) ????6:11
??????:&nbsp;"user-zh"<[email protected]&gt;;

????:&nbsp;Re: ????flink-jdbc-driver????mysql??????(flink1.10.0)



Hi,

???? batch ???????????? UpsertTableSink???????????? PR ????????????
https://issues.apache.org/jira/browse/FLINK-15579

Best,
Jark

On Tue, 25 Feb 2020 at 11:13, ???? <[email protected]&gt; wrote:

&gt; 
??????????jdbc????gateway????????mysql??????????????????jdbc-driver????mysql??????????????????????????????SQL-gateway??????????????????????????sqlclient??????????????????????????????????????????????????????????????????????????????????????tablesink??????????????????Caused
&gt; by: org.apache.flink.table.api.TableException: RetractStreamTableSink and
&gt; UpsertStreamTableSink is not supported in Batch 
environment.&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; at
&gt; 
org.apache.flink.table.planner.plan.nodes.physical.batch.BatchExecSink.translateToPlanInternal(BatchExecSink.scala:85)
&gt;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; at
&gt; 
org.apache.flink.table.planner.plan.nodes.physical.batch.BatchExecSink.translateToPlanInternal(BatchExecSink.scala:48)
&gt;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; at
&gt; 
org.apache.flink.table.planner.plan.nodes.exec.ExecNode$class.translateToPlan(ExecNode.scala:58)
&gt;&nbsp;&nbsp;&nbsp; at
&gt; 
org.apache.flink.table.planner.plan.nodes.physical.batch.BatchExecSink.translateToPlan(BatchExecSink.scala:48)
&gt;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; at
&gt; 
org.apache.flink.table.planner.delegation.BatchPlanner$$anonfun$translateToPlan$1.apply(BatchPlanner.scala:69)
&gt;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; at
&gt; 
org.apache.flink.table.planner.delegation.BatchPlanner$$anonfun$translateToPlan$1.apply(BatchPlanner.scala:68)
&gt; ????????????Connection connection =
&gt; DriverManager.getConnection("jdbc:flink://dataflow1:8083?planner=blink");
&gt; Statement statement = connection.createStatement();
&gt; sql = "CREATE TABLE t_user_target (\n" +
&gt;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; "&nbsp; id BIGINT,\n" +
&gt;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; "&nbsp; username 
VARCHAR\n" +
&gt;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; ") WITH (\n" +
&gt;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; "&nbsp; 'connector.type' = 
'jdbc',\n" +
&gt;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; "&nbsp; 'connector.driver' 
= 'com.mysql.jdbc.Driver',\n" +
&gt;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; "&nbsp; 'connector.url' = 
'jdbc:mysql://172.18.100.85:3306/targetdb',\n"
&gt; +
&gt;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; "&nbsp; 'connector.table' 
= 't1target',\n" +
&gt;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; "&nbsp; 
'connector.username' = 'root',\n" +
&gt;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; "&nbsp; 
'connector.password' = 'root',\n" +
&gt;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; "&nbsp; 
'connector.write.flush.max-rows' = '5000'\n" +
&gt;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; ");";
&gt; statement.executeUpdate(sql);
&gt; statement.execute("insert into t_user_target values(1,'fan'),(2,'ss')");

回复