使用flink-jdbc-driver写入mysql时失败(flink1.10.0)

classic Classic list List threaded Threaded
6 messages Options
Reply | Threaded
Open this post in threaded view
|

使用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')");
Reply | Threaded
Open this post in threaded view
|

Re: 使用flink-jdbc-driver写入mysql时失败(flink1.10.0)

Jark
Administrator
Hi,

当前 batch 模式还不支持 UpsertTableSink,不过已经有 PR 在支持中了:
https://issues.apache.org/jira/browse/FLINK-15579

Best,
Jark

On Tue, 25 Feb 2020 at 11:13, 猫猫 <[hidden email]> 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')");
Reply | Threaded
Open this post in threaded view
|

回复: 使用flink-jdbc-driver写入mysql时失败(flink1.10.0)

猫猫
感谢,还有一个疑问


流模式下和批模式下,写入mysql都是自动选用UpsertStreamTableSink吗?
我改为流模式运行,数据正确写入了,这两种模式在运行时有什么区别?


另外,如果基于批模式,是否目前没有可以执行insert into t_user_target values(1,'fan'),(2,'ss')的sink。
或者说,虽然存在批模式的sink,但是通过sql语句无法显示指定对应的sink。


------------------&nbsp;原始邮件&nbsp;------------------
发件人:&nbsp;"Jark Wu"<[hidden email]&gt;;
发送时间:&nbsp;2020年2月25日(星期二) 晚上6:11
收件人:&nbsp;"user-zh"<[hidden email]&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, 猫猫 <[hidden email]&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')");
Reply | Threaded
Open this post in threaded view
|

Re: 使用flink-jdbc-driver写入mysql时失败(flink1.10.0)

Jark
Administrator
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, 猫猫 <[hidden email]> wrote:

> 感谢,还有一个疑问
>
>
> 流模式下和批模式下,写入mysql都是自动选用UpsertStreamTableSink吗?
> 我改为流模式运行,数据正确写入了,这两种模式在运行时有什么区别?
>
>
> 另外,如果基于批模式,是否目前没有可以执行insert into t_user_target
> values(1,'fan'),(2,'ss')的sink。
> 或者说,虽然存在批模式的sink,但是通过sql语句无法显示指定对应的sink。
>
>
> ------------------&nbsp;原始邮件&nbsp;------------------
> 发件人:&nbsp;"Jark Wu"<[hidden email]&gt;;
> 发送时间:&nbsp;2020年2月25日(星期二) 晚上6:11
> 收件人:&nbsp;"user-zh"<[hidden email]&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, 猫猫 <[hidden email]&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')");
Reply | Threaded
Open this post in threaded view
|

回复: 使用flink-jdbc-driver写入mysql时失败(flink1.10.0)

猫猫
我目前只需要通过flink-sql完成mysql的ETL过程即可,即能够执行插入语句,能够使用insert table select完成数据的ETL就满足需要了。
而ETL在概念上应该适用批模式,但我使用批模式提交时,flink自动使用了UpsertStreamTableSink导致写入报错,所以需要寻找解决方法。


从这个目的出发,我的问题有两个:
1.既然sink是流式的,那么我提交时sql指令时,使用流模式就好了。
&nbsp; 但我不确定使用流模式运行和批模式运行会导致不同的处理结果吗?例如出现异常时,流模式会不会自动重试,导致数据重复。或者数据传递完成后,流并不会关闭。
&nbsp; 我关心的是:我不得不使用流模式提交sql指令时,与批模式相比需要有哪些注意的事项。
&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp;更进一步,我后续基于mysql对kafka和hive等集成时,流模式会带使用上的困扰吗?


&nbsp; 我并不执着于批或者流,而是希望了解在执行sql指令这个层面,是否能够完成我的ETL目标。




2.这与问题1有关,如果批模式或流模式运行效果是一样的。那么我后续就只使用流模式提交任务。也就没有下面的问题了。
&nbsp; 如果效果是不同的,那么创建表时,并没有选项让我选择批模式下能用的TableSink。
&nbsp; 我使用批模式时,系统也会选择UpsertStreamTableSink。这样运行就会报错,那么这是一个功能缺失吗?即flink-sql在批模式下没有正确的选择mysql的写入sink。
&nbsp; 那么我能够在sql定义层面去指定批写入的sink吗?或者要等该功能的完善?




------------------&nbsp;原始邮件&nbsp;------------------
发件人:&nbsp;"Jark Wu"<[hidden email]&gt;;
发送时间:&nbsp;2020年2月25日(星期二) 晚上7:28
收件人:&nbsp;"user-zh"<[hidden email]&gt;;

主题:&nbsp;Re: 使用flink-jdbc-driver写入mysql时失败(flink1.10.0)



Hi,

&gt; 流模式下和批模式下,写入mysql都是自动选用UpsertStreamTableSink吗?
是因为flink-jdbc目前只提供了 UpsertStreamTableSink 的实现。在如果是 StreamTableSink
或者说 AppendStreamTableSink 的话,批模式是支持的。


&gt; 另外,如果基于批模式,是否目前没有可以执行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, 猫猫 <[hidden email]&gt; wrote:

&gt; 感谢,还有一个疑问
&gt;
&gt;
&gt; 流模式下和批模式下,写入mysql都是自动选用UpsertStreamTableSink吗?
&gt; 我改为流模式运行,数据正确写入了,这两种模式在运行时有什么区别?
&gt;
&gt;
&gt; 另外,如果基于批模式,是否目前没有可以执行insert into t_user_target
&gt; values(1,'fan'),(2,'ss')的sink。
&gt; 或者说,虽然存在批模式的sink,但是通过sql语句无法显示指定对应的sink。
&gt;
&gt;
&gt; ------------------&amp;nbsp;原始邮件&amp;nbsp;------------------
&gt; 发件人:&amp;nbsp;"Jark Wu"<[hidden email]&amp;gt;;
&gt; 发送时间:&amp;nbsp;2020年2月25日(星期二) 晚上6:11
&gt; 收件人:&amp;nbsp;"user-zh"<[hidden email]&amp;gt;;
&gt;
&gt; 主题:&amp;nbsp;Re: 使用flink-jdbc-driver写入mysql时失败(flink1.10.0)
&gt;
&gt;
&gt;
&gt; Hi,
&gt;
&gt; 当前 batch 模式还不支持 UpsertTableSink,不过已经有 PR 在支持中了:
&gt; https://issues.apache.org/jira/browse/FLINK-15579
&gt;
&gt; Best,
&gt; Jark
&gt;
&gt; On Tue, 25 Feb 2020 at 11:13, 猫猫 <[hidden email]&amp;gt; wrote:
&gt;
&gt; &amp;gt;
&gt; 意图:通过jdbc访问gateway,做一个mysql表写入的测试。通过jdbc-driver创建mysql表成功,但执行写入数据时失败。SQL-gateway采用默认配置。但语句直接在sqlclient中执行可以成功。想请教一下,是定义不正确?还是环境配置不正确。正常来说不是应该默认使用tablesink吗?错误提示如下:Caused
&gt; &amp;gt; by: org.apache.flink.table.api.TableException: RetractStreamTableSink
&gt; and
&gt; &amp;gt; UpsertStreamTableSink is not supported in Batch
&gt; environment.&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp; at
&gt; &amp;gt;
&gt; org.apache.flink.table.planner.plan.nodes.physical.batch.BatchExecSink.translateToPlanInternal(BatchExecSink.scala:85)
&gt; &amp;gt;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp; at
&gt; &amp;gt;
&gt; org.apache.flink.table.planner.plan.nodes.physical.batch.BatchExecSink.translateToPlanInternal(BatchExecSink.scala:48)
&gt; &amp;gt;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp; at
&gt; &amp;gt;
&gt; org.apache.flink.table.planner.plan.nodes.exec.ExecNode$class.translateToPlan(ExecNode.scala:58)
&gt; &amp;gt;&amp;nbsp;&amp;nbsp;&amp;nbsp; at
&gt; &amp;gt;
&gt; org.apache.flink.table.planner.plan.nodes.physical.batch.BatchExecSink.translateToPlan(BatchExecSink.scala:48)
&gt; &amp;gt;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp; at
&gt; &amp;gt;
&gt; org.apache.flink.table.planner.delegation.BatchPlanner$$anonfun$translateToPlan$1.apply(BatchPlanner.scala:69)
&gt; &amp;gt;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp; at
&gt; &amp;gt;
&gt; org.apache.flink.table.planner.delegation.BatchPlanner$$anonfun$translateToPlan$1.apply(BatchPlanner.scala:68)
&gt; &amp;gt; 源代码如下:Connection connection =
&gt; &amp;gt;
&gt; DriverManager.getConnection("jdbc:flink://dataflow1:8083?planner=blink");
&gt; &amp;gt; Statement statement = connection.createStatement();
&gt; &amp;gt; sql = "CREATE TABLE t_user_target (\n" +
&gt; &amp;gt;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp; "&amp;nbsp; id BIGINT,\n"
&gt; +
&gt; &amp;gt;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp; "&amp;nbsp; username
&gt; VARCHAR\n" +
&gt; &amp;gt;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp; ") WITH (\n" +
&gt; &amp;gt;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp; "&amp;nbsp;
&gt; 'connector.type' = 'jdbc',\n" +
&gt; &amp;gt;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp; "&amp;nbsp;
&gt; 'connector.driver' = 'com.mysql.jdbc.Driver',\n" +
&gt; &amp;gt;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp; "&amp;nbsp;
&gt; 'connector.url' = 'jdbc:mysql://172.18.100.85:3306/targetdb',\n"
&gt; &amp;gt; +
&gt; &amp;gt;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp; "&amp;nbsp;
&gt; 'connector.table' = 't1target',\n" +
&gt; &amp;gt;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp; "&amp;nbsp;
&gt; 'connector.username' = 'root',\n" +
&gt; &amp;gt;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp; "&amp;nbsp;
&gt; 'connector.password' = 'root',\n" +
&gt; &amp;gt;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp; "&amp;nbsp;
&gt; 'connector.write.flush.max-rows' = '5000'\n" +
&gt; &amp;gt;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp; ");";
&gt; &amp;gt; statement.executeUpdate(sql);
&gt; &amp;gt; statement.execute("insert into t_user_target
&gt; values(1,'fan'),(2,'ss')");
Reply | Threaded
Open this post in threaded view
|

Re: 使用flink-jdbc-driver写入mysql时失败(flink1.10.0)

Jark
Administrator
Hi,

> 流模式和批模式的区别:
批模式会等所有数据都处理完以后,一次性输出最终结果。流模式会持续不断输出结果,以及更新结果,是一种增量模式。
对于相同的数据集以及相同的 query,从最终结果上来说,流与批运行的结果是一致的。
但是批模式必然要比流模式性能更优(因为计算量更少),流模式的优势是"提前(实时)看到结果"。
如果你后续想跟 hive 集成,那么批模式会更自然些。

个人建议你等一下这个 PR: https://github.com/apache/flink/pull/11045, 估计这两天就能 merge,等
merge 后你可以基于 master 重新打一个包,你的问题就解决了。

Best,
Jark

On Wed, 26 Feb 2020 at 08:26, 猫猫 <[hidden email]> wrote:

> 我目前只需要通过flink-sql完成mysql的ETL过程即可,即能够执行插入语句,能够使用insert table
> select完成数据的ETL就满足需要了。
>
> 而ETL在概念上应该适用批模式,但我使用批模式提交时,flink自动使用了UpsertStreamTableSink导致写入报错,所以需要寻找解决方法。
>
>
> 从这个目的出发,我的问题有两个:
> 1.既然sink是流式的,那么我提交时sql指令时,使用流模式就好了。
> &nbsp;
> 但我不确定使用流模式运行和批模式运行会导致不同的处理结果吗?例如出现异常时,流模式会不会自动重试,导致数据重复。或者数据传递完成后,流并不会关闭。
> &nbsp; 我关心的是:我不得不使用流模式提交sql指令时,与批模式相比需要有哪些注意的事项。
> &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp;
> &nbsp;更进一步,我后续基于mysql对kafka和hive等集成时,流模式会带使用上的困扰吗?
>
>
> &nbsp; 我并不执着于批或者流,而是希望了解在执行sql指令这个层面,是否能够完成我的ETL目标。
>
>
>
>
> 2.这与问题1有关,如果批模式或流模式运行效果是一样的。那么我后续就只使用流模式提交任务。也就没有下面的问题了。
> &nbsp; 如果效果是不同的,那么创建表时,并没有选项让我选择批模式下能用的TableSink。
> &nbsp;
> 我使用批模式时,系统也会选择UpsertStreamTableSink。这样运行就会报错,那么这是一个功能缺失吗?即flink-sql在批模式下没有正确的选择mysql的写入sink。
> &nbsp; 那么我能够在sql定义层面去指定批写入的sink吗?或者要等该功能的完善?
>
>
>
>
> ------------------&nbsp;原始邮件&nbsp;------------------
> 发件人:&nbsp;"Jark Wu"<[hidden email]&gt;;
> 发送时间:&nbsp;2020年2月25日(星期二) 晚上7:28
> 收件人:&nbsp;"user-zh"<[hidden email]&gt;;
>
> 主题:&nbsp;Re: 使用flink-jdbc-driver写入mysql时失败(flink1.10.0)
>
>
>
> Hi,
>
> &gt; 流模式下和批模式下,写入mysql都是自动选用UpsertStreamTableSink吗?
> 是因为flink-jdbc目前只提供了 UpsertStreamTableSink 的实现。在如果是 StreamTableSink
> 或者说 AppendStreamTableSink 的话,批模式是支持的。
>
>
> &gt; 另外,如果基于批模式,是否目前没有可以执行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, 猫猫 <[hidden email]&gt; wrote:
>
> &gt; 感谢,还有一个疑问
> &gt;
> &gt;
> &gt; 流模式下和批模式下,写入mysql都是自动选用UpsertStreamTableSink吗?
> &gt; 我改为流模式运行,数据正确写入了,这两种模式在运行时有什么区别?
> &gt;
> &gt;
> &gt; 另外,如果基于批模式,是否目前没有可以执行insert into t_user_target
> &gt; values(1,'fan'),(2,'ss')的sink。
> &gt; 或者说,虽然存在批模式的sink,但是通过sql语句无法显示指定对应的sink。
> &gt;
> &gt;
> &gt; ------------------&amp;nbsp;原始邮件&amp;nbsp;------------------
> &gt; 发件人:&amp;nbsp;"Jark Wu"<[hidden email]&amp;gt;;
> &gt; 发送时间:&amp;nbsp;2020年2月25日(星期二) 晚上6:11
> &gt; 收件人:&amp;nbsp;"user-zh"<[hidden email]&amp;gt;;
> &gt;
> &gt; 主题:&amp;nbsp;Re: 使用flink-jdbc-driver写入mysql时失败(flink1.10.0)
> &gt;
> &gt;
> &gt;
> &gt; Hi,
> &gt;
> &gt; 当前 batch 模式还不支持 UpsertTableSink,不过已经有 PR 在支持中了:
> &gt; https://issues.apache.org/jira/browse/FLINK-15579
> &gt;
> &gt; Best,
> &gt; Jark
> &gt;
> &gt; On Tue, 25 Feb 2020 at 11:13, 猫猫 <[hidden email]&amp;gt; wrote:
> &gt;
> &gt; &amp;gt;
> &gt;
> 意图:通过jdbc访问gateway,做一个mysql表写入的测试。通过jdbc-driver创建mysql表成功,但执行写入数据时失败。SQL-gateway采用默认配置。但语句直接在sqlclient中执行可以成功。想请教一下,是定义不正确?还是环境配置不正确。正常来说不是应该默认使用tablesink吗?错误提示如下:Caused
> &gt; &amp;gt; by: org.apache.flink.table.api.TableException:
> RetractStreamTableSink
> &gt; and
> &gt; &amp;gt; UpsertStreamTableSink is not supported in Batch
> &gt;
> environment.&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;
> at
> &gt; &amp;gt;
> &gt;
> org.apache.flink.table.planner.plan.nodes.physical.batch.BatchExecSink.translateToPlanInternal(BatchExecSink.scala:85)
> &gt; &amp;gt;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp; at
> &gt; &amp;gt;
> &gt;
> org.apache.flink.table.planner.plan.nodes.physical.batch.BatchExecSink.translateToPlanInternal(BatchExecSink.scala:48)
> &gt; &amp;gt;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp; at
> &gt; &amp;gt;
> &gt;
> org.apache.flink.table.planner.plan.nodes.exec.ExecNode$class.translateToPlan(ExecNode.scala:58)
> &gt; &amp;gt;&amp;nbsp;&amp;nbsp;&amp;nbsp; at
> &gt; &amp;gt;
> &gt;
> org.apache.flink.table.planner.plan.nodes.physical.batch.BatchExecSink.translateToPlan(BatchExecSink.scala:48)
> &gt; &amp;gt;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp; at
> &gt; &amp;gt;
> &gt;
> org.apache.flink.table.planner.delegation.BatchPlanner$$anonfun$translateToPlan$1.apply(BatchPlanner.scala:69)
> &gt; &amp;gt;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp; at
> &gt; &amp;gt;
> &gt;
> org.apache.flink.table.planner.delegation.BatchPlanner$$anonfun$translateToPlan$1.apply(BatchPlanner.scala:68)
> &gt; &amp;gt; 源代码如下:Connection connection =
> &gt; &amp;gt;
> &gt;
> DriverManager.getConnection("jdbc:flink://dataflow1:8083?planner=blink");
> &gt; &amp;gt; Statement statement = connection.createStatement();
> &gt; &amp;gt; sql = "CREATE TABLE t_user_target (\n" +
> &gt;
> &amp;gt;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;
> "&amp;nbsp; id BIGINT,\n"
> &gt; +
> &gt;
> &amp;gt;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;
> "&amp;nbsp; username
> &gt; VARCHAR\n" +
> &gt;
> &amp;gt;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;
> ") WITH (\n" +
> &gt;
> &amp;gt;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;
> "&amp;nbsp;
> &gt; 'connector.type' = 'jdbc',\n" +
> &gt;
> &amp;gt;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;
> "&amp;nbsp;
> &gt; 'connector.driver' = 'com.mysql.jdbc.Driver',\n" +
> &gt;
> &amp;gt;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;
> "&amp;nbsp;
> &gt; 'connector.url' = 'jdbc:mysql://172.18.100.85:3306/targetdb',\n"
> &gt; &amp;gt; +
> &gt;
> &amp;gt;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;
> "&amp;nbsp;
> &gt; 'connector.table' = 't1target',\n" +
> &gt;
> &amp;gt;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;
> "&amp;nbsp;
> &gt; 'connector.username' = 'root',\n" +
> &gt;
> &amp;gt;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;
> "&amp;nbsp;
> &gt; 'connector.password' = 'root',\n" +
> &gt;
> &amp;gt;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;
> "&amp;nbsp;
> &gt; 'connector.write.flush.max-rows' = '5000'\n" +
> &gt;
> &amp;gt;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp;
> ");";
> &gt; &amp;gt; statement.executeUpdate(sql);
> &gt; &amp;gt; statement.execute("insert into t_user_target
> &gt; values(1,'fan'),(2,'ss')");