You are viewing a plain text version of this content. The canonical link for it is here.
Posted to user-zh@flink.apache.org by 猫猫 <16...@qq.com> on 2020/02/25 03:13:14 UTC
使用flink-jdbc-driver写入mysql时失败(flink1.10.0)
意图：通过jdbc访问gateway，做一个mysql表写入的测试。通过jdbc-driver创建mysql表成功，但执行写入数据时失败。SQL-gateway采用默认配置。但语句直接在sqlclient中执行可以成功。想请教一下，是定义不正确？还是环境配置不正确。正常来说不是应该默认使用tablesink吗？错误提示如下：Caused by: org.apache.flink.table.api.TableException: RetractStreamTableSink and UpsertStreamTableSink is not supported in Batch environment. at org.apache.flink.table.planner.plan.nodes.physical.batch.BatchExecSink.translateToPlanInternal(BatchExecSink.scala:85) at org.apache.flink.table.planner.plan.nodes.physical.batch.BatchExecSink.translateToPlanInternal(BatchExecSink.scala:48) at org.apache.flink.table.planner.plan.nodes.exec.ExecNode$class.translateToPlan(ExecNode.scala:58) at org.apache.flink.table.planner.plan.nodes.physical.batch.BatchExecSink.translateToPlan(BatchExecSink.scala:48) at org.apache.flink.table.planner.delegation.BatchPlanner$$anonfun$translateToPlan$1.apply(BatchPlanner.scala:69) at org.apache.flink.table.planner.delegation.BatchPlanner$$anonfun$translateToPlan$1.apply(BatchPlanner.scala:68)
源代码如下：Connection connection = DriverManager.getConnection("jdbc:flink://dataflow1:8083?planner=blink");
Statement statement = connection.createStatement();
sql = "CREATE TABLE t_user_target (\n" +
" id BIGINT,\n" +
" username VARCHAR\n" +
") WITH (\n" +
" 'connector.type' = 'jdbc',\n" +
" 'connector.driver' = 'com.mysql.jdbc.Driver',\n" +
" 'connector.url' = 'jdbc:mysql://172.18.100.85:3306/targetdb',\n" +
" 'connector.table' = 't1target',\n" +
" 'connector.username' = 'root',\n" +
" 'connector.password' = 'root',\n" +
" 'connector.write.flush.max-rows' = '5000'\n" +
");";
statement.executeUpdate(sql);
statement.execute("insert into t_user_target values(1,'fan'),(2,'ss')");
Re: 使用flink-jdbc-driver写入mysql时失败(flink1.10.0)
Posted by Jark Wu <im...@gmail.com>.
Hi,
> 流模式和批模式的区别：
批模式会等所有数据都处理完以后，一次性输出最终结果。流模式会持续不断输出结果，以及更新结果，是一种增量模式。
对于相同的数据集以及相同的 query，从最终结果上来说，流与批运行的结果是一致的。
但是批模式必然要比流模式性能更优（因为计算量更少），流模式的优势是"提前（实时）看到结果"。
如果你后续想跟 hive 集成，那么批模式会更自然些。
个人建议你等一下这个 PR: https://github.com/apache/flink/pull/11045, 估计这两天就能 merge，等
merge 后你可以基于 master 重新打一个包，你的问题就解决了。
Best,
Jark
On Wed, 26 Feb 2020 at 08:26, 猫猫 <16...@qq.com> wrote:
> 我目前只需要通过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"<imjark@gmail.com>;
> 发送时间: 2020年2月25日(星期二) 晚上7:28
> 收件人: "user-zh"<user-zh@flink.apache.org>;
>
> 主题: 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, 猫猫 <16770864@qq.com> wrote:
>
> > 感谢，还有一个疑问
> >
> >
> > 流模式下和批模式下，写入mysql都是自动选用UpsertStreamTableSink吗？
> > 我改为流模式运行，数据正确写入了，这两种模式在运行时有什么区别？
> >
> >
> > 另外，如果基于批模式，是否目前没有可以执行insert into t_user_target
> > values(1,'fan'),(2,'ss')的sink。
> > 或者说，虽然存在批模式的sink，但是通过sql语句无法显示指定对应的sink。
> >
> >
> > ------------------&nbsp;原始邮件&nbsp;------------------
> > 发件人:&nbsp;"Jark Wu"<imjark@gmail.com&gt;;
> > 发送时间:&nbsp;2020年2月25日(星期二) 晚上6:11
> > 收件人:&nbsp;"user-zh"<user-zh@flink.apache.org&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, 猫猫 <16770864@qq.com&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')");
回复： 使用flink-jdbc-driver写入mysql时失败(flink1.10.0)
Posted by 猫猫 <16...@qq.com>.
我目前只需要通过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"<imjark@gmail.com>;
发送时间: 2020年2月25日(星期二) 晚上7:28
收件人: "user-zh"<user-zh@flink.apache.org>;
主题: 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, 猫猫 <16770864@qq.com> wrote:
> 感谢，还有一个疑问
>
>
> 流模式下和批模式下，写入mysql都是自动选用UpsertStreamTableSink吗？
> 我改为流模式运行，数据正确写入了，这两种模式在运行时有什么区别？
>
>
> 另外，如果基于批模式，是否目前没有可以执行insert into t_user_target
> values(1,'fan'),(2,'ss')的sink。
> 或者说，虽然存在批模式的sink，但是通过sql语句无法显示指定对应的sink。
>
>
> ------------------&nbsp;原始邮件&nbsp;------------------
> 发件人:&nbsp;"Jark Wu"<imjark@gmail.com&gt;;
> 发送时间:&nbsp;2020年2月25日(星期二) 晚上6:11
> 收件人:&nbsp;"user-zh"<user-zh@flink.apache.org&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, 猫猫 <16770864@qq.com&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')");
Re: 使用flink-jdbc-driver写入mysql时失败(flink1.10.0)
Posted by Jark Wu <im...@gmail.com>.
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, 猫猫 <16...@qq.com> wrote:
> 感谢，还有一个疑问
>
>
> 流模式下和批模式下，写入mysql都是自动选用UpsertStreamTableSink吗？
> 我改为流模式运行，数据正确写入了，这两种模式在运行时有什么区别？
>
>
> 另外，如果基于批模式，是否目前没有可以执行insert into t_user_target
> values(1,'fan'),(2,'ss')的sink。
> 或者说，虽然存在批模式的sink，但是通过sql语句无法显示指定对应的sink。
>
>
> ------------------ 原始邮件 ------------------
> 发件人: "Jark Wu"<imjark@gmail.com>;
> 发送时间: 2020年2月25日(星期二) 晚上6:11
> 收件人: "user-zh"<user-zh@flink.apache.org>;
>
> 主题: 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, 猫猫 <16770864@qq.com> wrote:
>
> >
> 意图：通过jdbc访问gateway，做一个mysql表写入的测试。通过jdbc-driver创建mysql表成功，但执行写入数据时失败。SQL-gateway采用默认配置。但语句直接在sqlclient中执行可以成功。想请教一下，是定义不正确？还是环境配置不正确。正常来说不是应该默认使用tablesink吗？错误提示如下：Caused
> > by: org.apache.flink.table.api.TableException: RetractStreamTableSink
> and
> > UpsertStreamTableSink is not supported in Batch
> environment. at
> >
> org.apache.flink.table.planner.plan.nodes.physical.batch.BatchExecSink.translateToPlanInternal(BatchExecSink.scala:85)
> > at
> >
> org.apache.flink.table.planner.plan.nodes.physical.batch.BatchExecSink.translateToPlanInternal(BatchExecSink.scala:48)
> > at
> >
> org.apache.flink.table.planner.plan.nodes.exec.ExecNode$class.translateToPlan(ExecNode.scala:58)
> > at
> >
> org.apache.flink.table.planner.plan.nodes.physical.batch.BatchExecSink.translateToPlan(BatchExecSink.scala:48)
> > at
> >
> org.apache.flink.table.planner.delegation.BatchPlanner$$anonfun$translateToPlan$1.apply(BatchPlanner.scala:69)
> > at
> >
> org.apache.flink.table.planner.delegation.BatchPlanner$$anonfun$translateToPlan$1.apply(BatchPlanner.scala:68)
> > 源代码如下：Connection connection =
> >
> DriverManager.getConnection("jdbc:flink://dataflow1:8083?planner=blink");
> > Statement statement = connection.createStatement();
> > sql = "CREATE TABLE t_user_target (\n" +
> > " id BIGINT,\n"
> +
> > " username
> VARCHAR\n" +
> > ") WITH (\n" +
> > "
> 'connector.type' = 'jdbc',\n" +
> > "
> 'connector.driver' = 'com.mysql.jdbc.Driver',\n" +
> > "
> 'connector.url' = 'jdbc:mysql://172.18.100.85:3306/targetdb',\n"
> > +
> > "
> 'connector.table' = 't1target',\n" +
> > "
> 'connector.username' = 'root',\n" +
> > "
> 'connector.password' = 'root',\n" +
> > "
> 'connector.write.flush.max-rows' = '5000'\n" +
> > ");";
> > statement.executeUpdate(sql);
> > statement.execute("insert into t_user_target
> values(1,'fan'),(2,'ss')");
回复： 使用flink-jdbc-driver写入mysql时失败(flink1.10.0)
Posted by 猫猫 <16...@qq.com>.
感谢，还有一个疑问
流模式下和批模式下，写入mysql都是自动选用UpsertStreamTableSink吗？
我改为流模式运行，数据正确写入了，这两种模式在运行时有什么区别？
另外，如果基于批模式，是否目前没有可以执行insert into t_user_target values(1,'fan'),(2,'ss')的sink。
或者说，虽然存在批模式的sink，但是通过sql语句无法显示指定对应的sink。
------------------ 原始邮件 ------------------
发件人: "Jark Wu"<imjark@gmail.com>;
发送时间: 2020年2月25日(星期二) 晚上6:11
收件人: "user-zh"<user-zh@flink.apache.org>;
主题: 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, 猫猫 <16770864@qq.com> wrote:
> 意图：通过jdbc访问gateway，做一个mysql表写入的测试。通过jdbc-driver创建mysql表成功，但执行写入数据时失败。SQL-gateway采用默认配置。但语句直接在sqlclient中执行可以成功。想请教一下，是定义不正确？还是环境配置不正确。正常来说不是应该默认使用tablesink吗？错误提示如下：Caused
> by: org.apache.flink.table.api.TableException: RetractStreamTableSink and
> UpsertStreamTableSink is not supported in Batch environment. at
> org.apache.flink.table.planner.plan.nodes.physical.batch.BatchExecSink.translateToPlanInternal(BatchExecSink.scala:85)
> at
> org.apache.flink.table.planner.plan.nodes.physical.batch.BatchExecSink.translateToPlanInternal(BatchExecSink.scala:48)
> at
> org.apache.flink.table.planner.plan.nodes.exec.ExecNode$class.translateToPlan(ExecNode.scala:58)
> at
> org.apache.flink.table.planner.plan.nodes.physical.batch.BatchExecSink.translateToPlan(BatchExecSink.scala:48)
> at
> org.apache.flink.table.planner.delegation.BatchPlanner$$anonfun$translateToPlan$1.apply(BatchPlanner.scala:69)
> at
> org.apache.flink.table.planner.delegation.BatchPlanner$$anonfun$translateToPlan$1.apply(BatchPlanner.scala:68)
> 源代码如下：Connection connection =
> DriverManager.getConnection("jdbc:flink://dataflow1:8083?planner=blink");
> Statement statement = connection.createStatement();
> sql = "CREATE TABLE t_user_target (\n" +
> " id BIGINT,\n" +
> " username VARCHAR\n" +
> ") WITH (\n" +
> " 'connector.type' = 'jdbc',\n" +
> " 'connector.driver' = 'com.mysql.jdbc.Driver',\n" +
> " 'connector.url' = 'jdbc:mysql://172.18.100.85:3306/targetdb',\n"
> +
> " 'connector.table' = 't1target',\n" +
> " 'connector.username' = 'root',\n" +
> " 'connector.password' = 'root',\n" +
> " 'connector.write.flush.max-rows' = '5000'\n" +
> ");";
> statement.executeUpdate(sql);
> statement.execute("insert into t_user_target values(1,'fan'),(2,'ss')");
Re: 使用flink-jdbc-driver写入mysql时失败(flink1.10.0)
Posted by Jark Wu <im...@gmail.com>.
Hi,
当前 batch 模式还不支持 UpsertTableSink，不过已经有 PR 在支持中了：
https://issues.apache.org/jira/browse/FLINK-15579
Best,
Jark
On Tue, 25 Feb 2020 at 11:13, 猫猫 <16...@qq.com> wrote:
> 意图：通过jdbc访问gateway，做一个mysql表写入的测试。通过jdbc-driver创建mysql表成功，但执行写入数据时失败。SQL-gateway采用默认配置。但语句直接在sqlclient中执行可以成功。想请教一下，是定义不正确？还是环境配置不正确。正常来说不是应该默认使用tablesink吗？错误提示如下：Caused
> by: org.apache.flink.table.api.TableException: RetractStreamTableSink and
> UpsertStreamTableSink is not supported in Batch environment. at
> org.apache.flink.table.planner.plan.nodes.physical.batch.BatchExecSink.translateToPlanInternal(BatchExecSink.scala:85)
> at
> org.apache.flink.table.planner.plan.nodes.physical.batch.BatchExecSink.translateToPlanInternal(BatchExecSink.scala:48)
> at
> org.apache.flink.table.planner.plan.nodes.exec.ExecNode$class.translateToPlan(ExecNode.scala:58)
> at
> org.apache.flink.table.planner.plan.nodes.physical.batch.BatchExecSink.translateToPlan(BatchExecSink.scala:48)
> at
> org.apache.flink.table.planner.delegation.BatchPlanner$$anonfun$translateToPlan$1.apply(BatchPlanner.scala:69)
> at
> org.apache.flink.table.planner.delegation.BatchPlanner$$anonfun$translateToPlan$1.apply(BatchPlanner.scala:68)
> 源代码如下：Connection connection =
> DriverManager.getConnection("jdbc:flink://dataflow1:8083?planner=blink");
> Statement statement = connection.createStatement();
> sql = "CREATE TABLE t_user_target (\n" +
> " id BIGINT,\n" +
> " username VARCHAR\n" +
> ") WITH (\n" +
> " 'connector.type' = 'jdbc',\n" +
> " 'connector.driver' = 'com.mysql.jdbc.Driver',\n" +
> " 'connector.url' = 'jdbc:mysql://172.18.100.85:3306/targetdb',\n"
> +
> " 'connector.table' = 't1target',\n" +
> " 'connector.username' = 'root',\n" +
> " 'connector.password' = 'root',\n" +
> " 'connector.write.flush.max-rows' = '5000'\n" +
> ");";
> statement.executeUpdate(sql);
> statement.execute("insert into t_user_target values(1,'fan'),(2,'ss')");