????????????????flink-sql????mysql??ETL??????????????????????????????????????insert table select??????????ETL?????????????? ??ETL??????????????????????????????????????????????flink??????????UpsertStreamTableSink????????????????????????????????????
???????????????????????????????? 1.????sink??????????????????????sql?????????????????????????? ?????????????????????????????????????????????????????????????????????????????????????????????????????????????????????????????????????????????????? ??????????????????????????????????sql?????????????????????????????????????????? ????????????????????mysql??kafka??hive???????????????????????????????????? ????????????????????????????????????????sql??????????????????????????????ETL?????? 2.????????1?????????????????????????????????????????????????????????????????????????????????????????????????? ????????????????????????????????????????????????????????????????TableSink?? ????????????????????????????UpsertStreamTableSink??????????????????????????????????????????????flink-sql????????????????????????mysql??????sink?? ????????????sql??????????????????????sink?????????????????????????? ------------------ ???????? ------------------ ??????: "Jark Wu"<[email protected]>; ????????: 2020??2??25??(??????) ????7:28 ??????: "user-zh"<[email protected]>; ????: Re: ????flink-jdbc-driver????mysql??????(flink1.10.0) Hi, > ????????????????????????mysql????????????UpsertStreamTableSink???? ??????flink-jdbc???????????? UpsertStreamTableSink ???????????????? StreamTableSink ?????? AppendStreamTableSink ?????????????????????? > ??????????????????????????????????????????insert into t_user_target values(1,'fan'),(2,'ss')??sink?? ?????????? sink ???????????? hive sink??filesystem sink ?????? Best, Jark On Tue, 25 Feb 2020 at 18:20, ???? <[email protected]> wrote: > ?????????????????? > > > ????????????????????????mysql????????????UpsertStreamTableSink???? > ???????????????????????????????????????????????????????????????? > > > ??????????????????????????????????????????insert into t_user_target > values(1,'fan'),(2,'ss')??sink?? > ????????????????????????sink??????????sql??????????????????????sink?? > > > ------------------&nbsp;????????&nbsp;------------------ > ??????:&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')");
