官方pyflink 例子的执行问题
官方例子: https://flink.apache.org/2020/04/09/pyflink-udf-support-flink.html 按照例子写了程序,也安装了pyflink | python -m pip install apache-flink | 代码: | from pyflink.datastream import StreamExecutionEnvironment from pyflink.table import StreamTableEnvironment, DataTypes from pyflink.table.descriptors import Schema, OldCsv, FileSystem from pyflink.table.udf import udf env = StreamExecutionEnvironment.get_execution_environment() env.set_parallelism(1) t_env = StreamTableEnvironment.create(env) add = udf(lambda i, j: i + j, [DataTypes.BIGINT(), DataTypes.BIGINT()], DataTypes.BIGINT()) t_env.register_function("add", add) t_env.connect(FileSystem().path('C:/Users/xuyin/Desktop/docker_compose_test/src.txt')) \ .with_format(OldCsv() .field('a', DataTypes.BIGINT()) .field('b', DataTypes.BIGINT())) \ .with_schema(Schema() .field('a', DataTypes.BIGINT()) .field('b', DataTypes.BIGINT())) \ .create_temporary_table('mySource') t_env.connect(FileSystem().path('C:/Users/xuyin/Desktop/docker_compose_test/tar.txt')) \ .with_format(OldCsv() .field('sum', DataTypes.BIGINT())) \ .with_schema(Schema() .field('sum', DataTypes.BIGINT())) \ .create_temporary_table('mySink') t_env.from_path('mySource')\ .select("add(a, b)") \ .insert_into('mySink') t_env.execute("tutorial_job") | 执行: | python test_pyflink.py | 报错: | Traceback (most recent call last): File "C:\Users\xuyin\AppData\Local\Programs\Python\Python37\lib\site-packages\pyflink\util\exceptions.py", line 147, in deco return f(*a, **kw) File "C:\Users\xuyin\AppData\Local\Programs\Python\Python37\lib\site-packages\py4j\protocol.py", line 328, in get_return_value format(target_id, ".", name), value) py4j.protocol.Py4JJavaError: An error occurred while calling o2.execute. : org.apache.flink.table.api.TableException: The configured Task Off-Heap Memory 0 bytes is less than the least required Python worker Memory 79 mb. The Task Off-Heap Memory can be configured using the configuration key 'taskmanager.memory.task.off-heap.size'. at org.apache.flink.table.planner.plan.nodes.common.CommonPythonBase$class.checkPythonWorkerMemory(CommonPythonBase.scala:158) at org.apache.flink.table.planner.plan.nodes.common.CommonPythonBase$class.getMergedConfiguration(CommonPythonBase.scala:119) at org.apache.flink.table.planner.plan.nodes.common.CommonPythonBase$class.getConfig(CommonPythonBase.scala:102) at org.apache.flink.table.planner.plan.nodes.physical.stream.StreamExecPythonCalc.getConfig(StreamExecPythonCalc.scala:35) at org.apache.flink.table.planner.plan.nodes.physical.stream.StreamExecPythonCalc.translateToPlanInternal(StreamExecPythonCalc.scala:61) at org.apache.flink.table.planner.plan.nodes.physical.stream.StreamExecPythonCalc.translateToPlanInternal(StreamExecPythonCalc.scala:35) at org.apache.flink.table.planner.plan.nodes.exec.ExecNode$class.translateToPlan(ExecNode.scala:58) at org.apache.flink.table.planner.plan.nodes.physical.stream.StreamExecCalcBase.translateToPlan(StreamExecCalcBase.scala:38) at org.apache.flink.table.planner.plan.nodes.physical.stream.StreamExecLegacySink.translateToTransformation(StreamExecLegacySink.scala:158) at org.apache.flink.table.planner.plan.nodes.physical.stream.StreamExecLegacySink.translateToPlanInternal(StreamExecLegacySink.scala:106) at org.apache.flink.table.planner.plan.nodes.physical.stream.StreamExecLegacySink.translateToPlanInternal(StreamExecLegacySink.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.stream.StreamExecLegacySink.translateToPlan(StreamExecLegacySink.scala:48) at org.apache.flink.table.planner.delegation.StreamPlanner$$anonfun$translateToPlan$1.apply(StreamPlanner.scala:67) at org.apache.flink.table.planner.delegation.StreamPlanner$$anonfun$translateToPlan$1.apply(StreamPlanner.scala:66) at scala.collection.TraversableLike$$anonfun$map$1.apply(TraversableLike.scala:234) at scala.collection.TraversableLike$$anonfun$map$1.apply(TraversableLike.scala:234) at scala.collection.Iterator$class.foreach(Iterator.scala:891) at scala.collection.AbstractIterator.foreach(Iterator.scala:1334) at scala.collection.IterableLike$class.foreach(IterableLike.scala:72) at scala.collection.AbstractIterable.foreach(Iterable.scala:54) at scala.collection.TraversableLike$class.map(TraversableLike.scala:234) at scala.collection.AbstractTraversable.map(Traversable.scala:104) at org.apache.flink.table.planner.delegation.StreamPlanner.translateToPlan(StreamPlanner.scala:66) at org.apache.flink.table.planner.delegation.PlannerBase.translate(PlannerBase.scala:166) at org.apache.flink.table.api.internal.TableEnvironmentImpl.translate(TableEnvironmentImpl.java:1248) at org.apache.flink
Re:Re: 官方pyflink 例子的执行问题
你好 明白了,感谢 , 我文档没看清楚哈 在 2020-07-21 11:44:23,"Xingbo Huang" 写道: >Hi, >你需要添加配置,如果你没有使用RocksDB作为statebackend的话,你直接配置t_env.get_config().get_configuration().set_boolean("python.fn-execution.memory.managed", >True)就行,如果你用了的话,就需要配置off-heap >memory了,table_env.get_config().get_configuration().set_string("taskmanager.memory.task.off-heap.size", >'80m')。你可以参考文档上的例子,以及对应的note说明[1] > >[1] >https://ci.apache.org/projects/flink/flink-docs-release-1.11/dev/table/python/python_udfs.html#scalar-functions > >Best, >Xingbo > > >chenxuying 于2020年7月21日周二 上午11:36写道: > >> 官方例子: >> https://flink.apache.org/2020/04/09/pyflink-udf-support-flink.html >> 按照例子写了程序,也安装了pyflink >> | >> python -m pip install apache-flink >> | >> 代码: >> | >> from pyflink.datastream import StreamExecutionEnvironment >> from pyflink.table import StreamTableEnvironment, DataTypes >> from pyflink.table.descriptors import Schema, OldCsv, FileSystem >> from pyflink.table.udf import udf >> >> >> env = StreamExecutionEnvironment.get_execution_environment() >> env.set_parallelism(1) >> t_env = StreamTableEnvironment.create(env) >> >> >> add = udf(lambda i, j: i + j, [DataTypes.BIGINT(), DataTypes.BIGINT()], >> DataTypes.BIGINT()) >> >> >> t_env.register_function("add", add) >> >> >> t_env.connect(FileSystem().path('C:/Users/xuyin/Desktop/docker_compose_test/src.txt')) >> \ >> .with_format(OldCsv() >> .field('a', DataTypes.BIGINT()) >> .field('b', DataTypes.BIGINT())) \ >> .with_schema(Schema() >> .field('a', DataTypes.BIGINT()) >> .field('b', DataTypes.BIGINT())) \ >> .create_temporary_table('mySource') >> >> >> t_env.connect(FileSystem().path('C:/Users/xuyin/Desktop/docker_compose_test/tar.txt')) >> \ >> .with_format(OldCsv() >> .field('sum', DataTypes.BIGINT())) \ >> .with_schema(Schema() >> .field('sum', DataTypes.BIGINT())) \ >> .create_temporary_table('mySink') >> >> >> t_env.from_path('mySource')\ >> .select("add(a, b)") \ >> .insert_into('mySink') >> >> >> t_env.execute("tutorial_job") >> | >> >> 执行: >> >> | >> python test_pyflink.py >> | >> >> 报错: >> >> >> | >> Traceback (most recent call last): >> File >> "C:\Users\xuyin\AppData\Local\Programs\Python\Python37\lib\site-packages\pyflink\util\exceptions.py", >> line 147, in deco >> return f(*a, **kw) >> File >> "C:\Users\xuyin\AppData\Local\Programs\Python\Python37\lib\site-packages\py4j\protocol.py", >> line 328, in get_return_value >> format(target_id, ".", name), value) >> py4j.protocol.Py4JJavaError: An error occurred while calling o2.execute. >> : org.apache.flink.table.api.TableException: The configured Task Off-Heap >> Memory 0 bytes is less than the least required Python worker Memory 79 mb. >> The Task Off-Heap Memory can be configured using the configuration key >> 'taskmanager.memory.task.off-heap.size'. >> at >> org.apache.flink.table.planner.plan.nodes.common.CommonPythonBase$class.checkPythonWorkerMemory(CommonPythonBase.scala:158) >> at >> org.apache.flink.table.planner.plan.nodes.common.CommonPythonBase$class.getMergedConfiguration(CommonPythonBase.scala:119) >> at >> org.apache.flink.table.planner.plan.nodes.common.CommonPythonBase$class.getConfig(CommonPythonBase.scala:102) >> at >> org.apache.flink.table.planner.plan.nodes.physical.stream.StreamExecPythonCalc.getConfig(StreamExecPythonCalc.scala:35) >> at >> org.apache.flink.table.planner.plan.nodes.physical.stream.StreamExecPythonCalc.translateToPlanInternal(StreamExecPythonCalc.scala:61) >> at >> org.apache.flink.table.planner.plan.nodes.physical.stream.StreamExecPythonCalc.translateToPlanInternal(StreamExecPythonCalc.scala:35) >> at >> org.apache.flink.table.planner.plan.nodes.exec.ExecNode$class.translateToPlan(ExecNode.scala:58) >> at >> org.apache.flink.table.planner.plan.nodes.physical.stream.StreamExecCalcBase.translateToPlan(StreamExecCalcBase.scala:38) >> at >> org.apache.flink.table.planner.plan.nodes.physical.stream.StreamExecLegacySink.translateToTransformation(StreamExecLegacySink.scala:158) >> at >> org.apache.flink.table.planner.plan.nodes.physical.
flinksql kafka 插入mysql 的insert语法问题和唯一主键的问题
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()'. Supported form(s): '$SCALAR_QUERY()' 问题二 : 如果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'
Re:Re: flinksql kafka 插入mysql 的insert语法问题和唯一主键的问题
谢谢回答 使用新属性可以 成功修改记录 , 但是不太明白 "使用老的option参数还是需要根据query推导主键" 这里话是什么意思,需要怎么做 在 2020-07-31 16:46:41,"Leonard Xu" 写道: >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 写道: >> >> 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(> A, VARCHAR(2147483647) B)>)'. Supported form(s): >> '$SCALAR_QUERY()' >> >> >> 问题二 : 如果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' >> >> >> >
Re:Re: flinksql kafka 插入mysql 的insert语法问题和唯一主键的问题
hi ok,谢谢,懂了哈哈 在 2020-07-31 21:27:02,"Leonard Xu" 写道: >Hello > >> 在 2020年7月31日,21:13,chenxuying 写道: >> >> 但是不太明白 "使用老的option参数还是需要根据query推导主键" 这里话是什么意思,需要怎么做 > >简单来讲,如果使用的是老版本(1.10)的option参数,代码执行的路径就和1.10版本一样的,1.10版本里是不支持定义 PRIMARY KEY 的, >是通过用户的query来决定写入的模式是upsert 还是 append , 你可以看下1.10的文档关于用query 推导 写入模式的文档[1], >如果已经在用1.11了,1.10的文档可以不用看的。 > >在1.10里经常出现query 推导不出 key 导致无法做upsert写入的case, 在1.11里通过支持定义 PRIMARY >KEY,不会再有类似问题.1.11的文档参考[2]。 > >祝好 >Leonard > >[1] >https://ci.apache.org/projects/flink/flink-docs-master/dev/table/connect.html#jdbc-connector > ><https://ci.apache.org/projects/flink/flink-docs-master/dev/table/connect.html#jdbc-connector> >[2] >https://ci.apache.org/projects/flink/flink-docs-master/dev/table/connectors/jdbc.html#how-to-create-a-jdbc-table > ><https://ci.apache.org/projects/flink/flink-docs-master/dev/table/connectors/jdbc.html#how-to-create-a-jdbc-table>
flinksql insert overwrite 报错 INSERT OVERWRITE requires JDBC:MySQL DynamicTableSink to implement SupportsOverwrite interface
Hello 请问在flink 1.11.0文档[1]中发现可以使用时可以使用INSERTOVERWRITE... 尝试了一下,但是在执行语句"insert overwrite mysqlsink select a,cast(b as varchar) b from mySource"时报如下错误 Exception in thread "main" org.apache.flink.table.api.ValidationException: INSERT OVERWRITE requires JDBC:MySQL DynamicTableSink to implement SupportsOverwrite interface. 是得自定义connector吗,实现DynamicTableSink? 祝好 chenxuying [1] https://ci.apache.org/projects/flink/flink-docs-release-1.11/dev/table/sql/insert.html#syntax
Re:Re: flinksql insert overwrite 报错 INSERT OVERWRITE requires JDBC:MySQL DynamicTableSink to implement SupportsOverwrite interface
你好,我这边只是做一个尝试 , 因为之前使用了insert into + 主键做到更新db记录的操作 , 然后现在看到INSERT OVERWRITE的语法就试了一下 , 原来OVERWRITE目前只支持 Filesystem connector 和 Hive table 然后还想问下在使用insert into + 主键 时,如果主键存在 , 则在底层执行sql时是update语句是吗, 还是delete+update , 我们这边之前有个需求是update执行效率太低 , 然后需要用到replace into , 不知道flink是否支持 在 2020-08-02 09:48:04,"Leonard Xu" 写道: >Hi, > >这个错是因为JDBC connector 不支持INSERT OVERWRITE, 你看的文档是列出了目前 Flink SQL >支持的INSERT语法,但是不是所有的 connector 都支持 INSERT OVERWRITE, 目前支持的只有 Filesystem >connector 和 Hive table, 这些表一般不会有主键。其他connector 如 JDBC\ES\HBase 目前不支持 INSERT >OVERWRITE,现在 JDBC\ES\HBase connector都是支持upsert 插入的[1], >就是在connector 表上定义了PK,结果可以按照PK更新,对于DB类的系统应该都是可以满足业务需求的。 可以分享下需要INSERT >OVERWRITE到DB的场景吗? > >Best >Leonard >[1]https://ci.apache.org/projects/flink/flink-docs-release-1.11/dev/table/connectors/jdbc.html#key-handling > ><https://ci.apache.org/projects/flink/flink-docs-release-1.11/dev/table/connectors/jdbc.html#key-handling> > >> 在 2020年8月1日,19:20,chenxuying 写道: >> >> Hello >> 请问在flink 1.11.0文档[1]中发现可以使用时可以使用INSERTOVERWRITE... 尝试了一下,但是在执行语句"insert >> overwrite mysqlsink select a,cast(b as varchar) b from mySource"时报如下错误 >> Exception in thread "main" org.apache.flink.table.api.ValidationException: >> INSERT OVERWRITE requires JDBC:MySQL DynamicTableSink to implement >> SupportsOverwrite interface. >> 是得自定义connector吗,实现DynamicTableSink? >> >> >> 祝好 >> chenxuying >> [1] >> https://ci.apache.org/projects/flink/flink-docs-release-1.11/dev/table/sql/insert.html#syntax >
Re:Re: flinksql insert overwrite 报错 INSERT OVERWRITE requires JDBC:MySQL DynamicTableSink to implement SupportsOverwrite interface
谢谢, 明白了 在 2020-08-03 10:42:53,"Leonard Xu" 写道: >如果 DB 支持 upsert 语法,执行的是update, 如果不支持 upsert语法, 则是 delete + insert,MySQL 和 PG >都支持 upsert, 底层对应的sql语句是 > >Database Upsert Grammar >MySQL INSERT .. ON DUPLICATE KEY UPDATE .. >PostgreSQL INSERT .. ON CONFLICT .. DO UPDATE SET .. > >MySQL connector 不支持 replace into, 用的是 on duplicate key update. > >祝好 >Leonard >[1] >https://ci.apache.org/projects/flink/flink-docs-master/dev/table/connectors/jdbc.html#idempotent-writes > ><https://ci.apache.org/projects/flink/flink-docs-master/dev/table/connectors/jdbc.html#idempotent-writes> > > >> 在 2020年8月3日,10:33,chenxuying 写道: >> >> 你好,我这边只是做一个尝试 , 因为之前使用了insert into + 主键做到更新db记录的操作 , 然后现在看到INSERT >> OVERWRITE的语法就试了一下 , 原来OVERWRITE目前只支持 Filesystem connector 和 Hive table >> 然后还想问下在使用insert into + 主键 时,如果主键存在 , 则在底层执行sql时是update语句是吗, 还是delete+update >> , 我们这边之前有个需求是update执行效率太低 , 然后需要用到replace into , 不知道flink是否支持 >> >> >> >> >> >> >> >> >> >> >> >> >> >> >> >> >> >> 在 2020-08-02 09:48:04,"Leonard Xu" 写道: >>> Hi, >>> >>> 这个错是因为JDBC connector 不支持INSERT OVERWRITE, 你看的文档是列出了目前 Flink SQL >>> 支持的INSERT语法,但是不是所有的 connector 都支持 INSERT OVERWRITE, 目前支持的只有 Filesystem >>> connector 和 Hive table, 这些表一般不会有主键。其他connector 如 JDBC\ES\HBase 目前不支持 >>> INSERT OVERWRITE,现在 JDBC\ES\HBase connector都是支持upsert 插入的[1], >>> 就是在connector 表上定义了PK,结果可以按照PK更新,对于DB类的系统应该都是可以满足业务需求的。 可以分享下需要INSERT >>> OVERWRITE到DB的场景吗? >>> >>> Best >>> Leonard >>> [1]https://ci.apache.org/projects/flink/flink-docs-release-1.11/dev/table/connectors/jdbc.html#key-handling >>> >>> <https://ci.apache.org/projects/flink/flink-docs-release-1.11/dev/table/connectors/jdbc.html#key-handling> >>> >>>> 在 2020年8月1日,19:20,chenxuying 写道: >>>> >>>> Hello >>>> 请问在flink 1.11.0文档[1]中发现可以使用时可以使用INSERTOVERWRITE... 尝试了一下,但是在执行语句"insert >>>> overwrite mysqlsink select a,cast(b as varchar) b from mySource"时报如下错误 >>>> Exception in thread "main" org.apache.flink.table.api.ValidationException: >>>> INSERT OVERWRITE requires JDBC:MySQL DynamicTableSink to implement >>>> SupportsOverwrite interface. >>>> 是得自定义connector吗,实现DynamicTableSink? >>>> >>>> >>>> 祝好 >>>> chenxuying >>>> [1] >>>> https://ci.apache.org/projects/flink/flink-docs-release-1.11/dev/table/sql/insert.html#syntax >>> >
有界数据中batch和stream的区别
hi : flink table sql 1.11.0 在EnvironmentSettings中可以设置BatchMode或StreamingMode EnvironmentSettings environmentSettings = EnvironmentSettings.newInstance() //.inStreamingMode() .inBatchMode() .build(); 如果使用mysql作为source , 使用这两种模式都可以运行 , 效果都一样 , 感觉对批和流的理解还不够 , 不知道其中区别 , 不知道大佬们有没有例子可以比较容易理解 我的代码 EnvironmentSettings environmentSettings = EnvironmentSettings.newInstance() //.inStreamingMode() .inBatchMode() .build(); TableEnvironment tableEnvironment = TableEnvironment.create(environmentSettings); tableEnvironment.executeSql("CREATE TABLE mysql_source ( " + " id bigint, " + " game_id varchar, " + " PRIMARY KEY (id) NOT ENFORCED " + " ) " + " with ( " + "'connector' = 'jdbc', " + " 'url' = 'jdbc:mysql://47.99.181.86:3306/flinksql?useSSL=false' , " + " 'username' = 'root' , " + " 'password' = 'root', " + " 'table-name' = 'mysqlsink' , " + " 'driver' = 'com.mysql.cj.jdbc.Driver' , " + " 'sink.buffer-flush.interval' = '2s', " + " 'sink.buffer-flush.max-rows' = '300' " + " )"); tableEnvironment.executeSql("CREATE TABLE print_sink ( " + " id bigint, " + " game_id varchar, " + " PRIMARY KEY (id) NOT ENFORCED " + " ) " + " with ( " + "'connector' = 'print' " + " )"); tableEnvironment.executeSql("insert into print_sink select id,game_id from mysql_source");
Re:Re: 有界数据中batch和stream的区别
你好,请问下我修改后的语句是 insert into print_sink select game_id,count(id) from mysql_source group by game_id 然后在执行的时候如果选择的是streamMode他会打印出Changelog,如下 2> +I(12,1) 5> +I(12555,1) 1> +I(122,1) 3> +I(13,1) 6> +I(1,1) 6> -U(1,1) 6> +U(1,2) 6> -U(1,2) 6> +U(1,3) 6> -U(1,3) 6> +U(1,4) 6> -U(1,4) 然后如果我使用的是batchMode,他就报错了 org.apache.flink.util.FlinkException: Error while shutting the TaskExecutor down. at org.apache.flink.runtime.taskexecutor.TaskExecutor.handleOnStopException(TaskExecutor.java:440) at org.apache.flink.runtime.taskexecutor.TaskExecutor.lambda$onStop$2(TaskExecutor.java:425) ... Caused by: java.util.concurrent.CompletionException: java.lang.NoClassDefFoundError: Could not initialize class org.apache.flink.util.JavaGcCleanerWrapper at java.util.concurrent.CompletableFuture.encodeThrowable(CompletableFuture.java:273) ... at org.apache.flink.runtime.taskexecutor.TaskExecutor.onStop(TaskExecutor.java:422) ... 21 more Suppressed: org.apache.flink.util.FlinkException: Could not properly shut down the TaskManager services. at org.apache.flink.runtime.taskexecutor.TaskManagerServices.shutDown(TaskManagerServices.java:236) at org.apache.flink.runtime.taskexecutor.TaskExecutor.stopTaskExecutorServices(TaskExecutor.java:462) at... ... 21 more Caused by: org.apache.flink.util.FlinkException: Could not close resource. at org.apache.flink.util.AutoCloseableAsync.close(AutoCloseableAsync.java:42) at org.apache.flink.runtime.taskexecutor.TaskManagerServices.shutDown(TaskManagerServices.java:204) ... 37 more Caused by: java.lang.NoClassDefFoundError: Could not initialize class org.apache.flink.util.JavaGcCleanerWrapper at org.apache.flink.runtime.memory.UnsafeMemoryBudget.reserveMemory(UnsafeMemoryBudget.java:94) at org.apache.flink.runtime.memory.UnsafeMemoryBudget.verifyEmpty(UnsafeMemoryBudget.java:64) ... at org.apache.flink.runtime.taskexecutor.TaskExecutor.onStop(TaskExecutor.java:422) ... 21 more [CIRCULAR REFERENCE:java.lang.NoClassDefFoundError: Could not initialize class org.apache.flink.util.JavaGcCleanerWrapper] 不知道您是否知道原因 在 2020-08-04 12:11:32,"godfrey he" 写道: >逻辑上批产生的结果是Table,流产生的结果是Changelog。 >你在例子中Table的结果和changelog的结果是一样的,所以你感觉差不多。 >最简单的方式可以将query改为带group by的,再看结果的差异。 >更多关于Table和Changelog的概念可以参考 [1] > >[1] >https://ci.apache.org/projects/flink/flink-docs-release-1.11/zh/dev/table/streaming/dynamic_tables.html > >chenxuying 于2020年8月4日周二 上午11:44写道: > >> hi : >> flink table sql 1.11.0 >> 在EnvironmentSettings中可以设置BatchMode或StreamingMode >> >> >> EnvironmentSettings environmentSettings = EnvironmentSettings.newInstance() >> //.inStreamingMode() >> .inBatchMode() >> .build(); >> >> >> 如果使用mysql作为source , 使用这两种模式都可以运行 , 效果都一样 , 感觉对批和流的理解还不够 , 不知道其中区别 , >> 不知道大佬们有没有例子可以比较容易理解 >> 我的代码 >> EnvironmentSettings environmentSettings = EnvironmentSettings.newInstance() >> //.inStreamingMode() >> .inBatchMode() >> .build(); >> TableEnvironment tableEnvironment = >> TableEnvironment.create(environmentSettings); >> tableEnvironment.executeSql("CREATE TABLE mysql_source ( " + >> " id bigint, " + >> " game_id varchar, " + >> " PRIMARY KEY (id) NOT ENFORCED " + >> " ) " + >> " with ( " + >> "'connector' = 'jdbc', " + >> " 'url' = 'jdbc:mysql://47.99.181.86:3306/flinksql?useSSL=false' , " + >> " 'username' = 'root' , " + >> " 'password' = 'root', " + >> " 'table-name' = 'mysqlsink' , " + >> " 'driver' = 'com.mysql.cj.jdbc.Driver' , " + >> " 'sink.buffer-flush.interval' = '2s', " + >> " 'sink.buffer-flush.max-rows' = '300' " + >> " )"); >> tableEnvironment.executeSql("CREATE TABLE print_sink ( " + >> " id bigint, " + >> " game_id varchar, " + >> " PRIMARY KEY (id) NOT ENFORCED " + >> " ) " + >> " with ( " + >> "'connector' = 'print' " + >> " )"); >> tableEnvironment.executeSql("insert into print_sink select id,game_id from >> mysql_source");
Re:Re: Re: 有界数据中batch和stream的区别
我的是在win10+idea上开发测试的, 然后同一个项目代码, 在我同事电脑上却可以正常运行, 不知道是不是系统 在 2020-08-04 17:19:48,"godfrey he" 写道: >你的运行环境是啥?能提供一下相关配置吗? > >chenxuying 于2020年8月4日周二 下午2:46写道: > >> 你好,请问下我修改后的语句是 >> insert into print_sink select game_id,count(id) from mysql_source group by >> game_id >> 然后在执行的时候如果选择的是streamMode他会打印出Changelog,如下 >> 2> +I(12,1) >> 5> +I(12555,1) 1> +I(122,1) 3> +I(13,1) 6> +I(1,1) 6> -U(1,1) 6> +U(1,2) >> 6> -U(1,2) 6> +U(1,3) 6> -U(1,3) 6> +U(1,4) 6> -U(1,4) >> >> >> 然后如果我使用的是batchMode,他就报错了 >> org.apache.flink.util.FlinkException: Error while shutting the >> TaskExecutor down. >> at >> org.apache.flink.runtime.taskexecutor.TaskExecutor.handleOnStopException(TaskExecutor.java:440) >> at >> org.apache.flink.runtime.taskexecutor.TaskExecutor.lambda$onStop$2(TaskExecutor.java:425) >> ... >> Caused by: java.util.concurrent.CompletionException: >> java.lang.NoClassDefFoundError: Could not initialize class >> org.apache.flink.util.JavaGcCleanerWrapper >> at >> java.util.concurrent.CompletableFuture.encodeThrowable(CompletableFuture.java:273) >> ... >> at >> org.apache.flink.runtime.taskexecutor.TaskExecutor.onStop(TaskExecutor.java:422) >> ... 21 more >> Suppressed: org.apache.flink.util.FlinkException: Could not properly shut >> down the TaskManager services. >> at >> org.apache.flink.runtime.taskexecutor.TaskManagerServices.shutDown(TaskManagerServices.java:236) >> at >> org.apache.flink.runtime.taskexecutor.TaskExecutor.stopTaskExecutorServices(TaskExecutor.java:462) >> at... >> ... 21 more >> Caused by: org.apache.flink.util.FlinkException: Could not close resource. >> at >> org.apache.flink.util.AutoCloseableAsync.close(AutoCloseableAsync.java:42) >> at >> org.apache.flink.runtime.taskexecutor.TaskManagerServices.shutDown(TaskManagerServices.java:204) >> ... 37 more >> Caused by: java.lang.NoClassDefFoundError: Could not initialize class >> org.apache.flink.util.JavaGcCleanerWrapper >> at >> org.apache.flink.runtime.memory.UnsafeMemoryBudget.reserveMemory(UnsafeMemoryBudget.java:94) >> at >> org.apache.flink.runtime.memory.UnsafeMemoryBudget.verifyEmpty(UnsafeMemoryBudget.java:64) >> ... >> at >> org.apache.flink.runtime.taskexecutor.TaskExecutor.onStop(TaskExecutor.java:422) >> ... 21 more >> [CIRCULAR REFERENCE:java.lang.NoClassDefFoundError: Could not initialize >> class org.apache.flink.util.JavaGcCleanerWrapper] >> >> >> 不知道您是否知道原因 >> >> >> 在 2020-08-04 12:11:32,"godfrey he" 写道: >> >逻辑上批产生的结果是Table,流产生的结果是Changelog。 >> >你在例子中Table的结果和changelog的结果是一样的,所以你感觉差不多。 >> >最简单的方式可以将query改为带group by的,再看结果的差异。 >> >更多关于Table和Changelog的概念可以参考 [1] >> > >> >[1] >> > >> https://ci.apache.org/projects/flink/flink-docs-release-1.11/zh/dev/table/streaming/dynamic_tables.html >> > >> >chenxuying 于2020年8月4日周二 上午11:44写道: >> > >> >> hi : >> >> flink table sql 1.11.0 >> >> 在EnvironmentSettings中可以设置BatchMode或StreamingMode >> >> >> >> >> >> EnvironmentSettings environmentSettings = >> EnvironmentSettings.newInstance() >> >> //.inStreamingMode() >> >> .inBatchMode() >> >> .build(); >> >> >> >> >> >> 如果使用mysql作为source , 使用这两种模式都可以运行 , 效果都一样 , 感觉对批和流的理解还不够 , 不知道其中区别 , >> >> 不知道大佬们有没有例子可以比较容易理解 >> >> 我的代码 >> >> EnvironmentSettings environmentSettings = >> EnvironmentSettings.newInstance() >> >> //.inStreamingMode() >> >> .inBatchMode() >> >> .build(); >> >> TableEnvironment tableEnvironment = >> >> TableEnvironment.create(environmentSettings); >> >> tableEnvironment.executeSql("CREATE TABLE mysql_source ( " + >> >> " id bigint, " + >> >> " game_id varchar, " + >> >> " PRIMARY KEY (id) NOT ENFORCED " + >> >> " ) " + >> >> " with ( " + >> >> "'connector' = 'jdbc', " + >> >> " 'url' = 'jdbc:mysql://47.99.181.86:3306/flinksql?useSSL=false' , " + >> >> " 'username' = 'root' , " + >> >> " 'password' = 'root', " + >> >> " 'table-name' = 'mysqlsink' , " + >> >> " 'driver' = 'com.mysql.cj.jdbc.Driver' , " + >> >> " 'sink.buffer-flush.interval' = '2s', " + >> >> " 'sink.buffer-flush.max-rows' = '300' " + >> >> " )"); >> >> tableEnvironment.executeSql("CREATE TABLE print_sink ( " + >> >> " id bigint, " + >> >> " game_id varchar, " + >> >> " PRIMARY KEY (id) NOT ENFORCED " + >> >> " ) " + >> >> " with ( " + >> >> "'connector' = 'print' " + >> >> " )"); >> >> tableEnvironment.executeSql("insert into print_sink select id,game_id >> from >> >> mysql_source"); >>
请问在flinksql中如何使用聚合函数 LISTAGG
版本: flinksql 1.11.0 需求: 需要实现多行聚合成一行功能 代码如下: environmentSettings = EnvironmentSettings.new_instance().in_streaming_mode().build() t_env = StreamTableEnvironment.create(environment_settings = environmentSettings) t_env.get_config().get_configuration().set_string("python.fn-execution.memory.managed", 'true') a_df = pd.DataFrame({"id":["1","1","1"],"uuid":["1","2","3"]}) a_table = t_env.from_pandas(a_df, DataTypes.ROW([DataTypes.FIELD("id", DataTypes.STRING()), DataTypes.FIELD("uuid", DataTypes.STRING())])) t_env.create_temporary_view("table_a",a_table) b_df = pd.DataFrame({"val":["3","4","5","6"],"uuid":["1","2","3","4"]}) table_b = t_env.from_pandas(b_df , DataTypes.ROW([DataTypes.FIELD("val", DataTypes.STRING()), DataTypes.FIELD("uuid", DataTypes.STRING())])) t_env.create_temporary_view("table_b",table_b) t_env.sql_update(""" CREATE TABLE mySink ( b varchar , c varchar ) WITH ( 'connector' = 'print' ) """) t_env.sql_update(""" insert into mySink select t1.id ,LISTAGG(t2.val , ',') from table_a t1 left join table_b t2 on t1.uuid = t2.uuid group by t1.id """) t_env.execute("tutorial_job") 报错: Caused by: java.lang.ClassCastException: org.apache.flink.table.data.binary.BinaryRawValueData cannot be cast to org.apache.flink.table.data.StringData at org.apache.flink.table.data.GenericRowData.getString(GenericRowData.java:169) at org.apache.flink.table.data.JoinedRowData.getString(JoinedRowData.java:139) at org.apache.flink.table.data.RowData.get(RowData.java:273) at org.apache.flink.table.runtime.typeutils.RowDataSerializer.copyRowData(RowDataSerializer.java:156) at org.apache.flink.table.runtime.typeutils.RowDataSerializer.copy(RowDataSerializer.java:123) at org.apache.flink.table.runtime.typeutils.RowDataSerializer.copy(RowDataSerializer.java:50) at org.apache.flink.streaming.runtime.tasks.OperatorChain$CopyingChainingOutput.pushToOperator(OperatorChain.java:715) at org.apache.flink.streaming.runtime.tasks.OperatorChain$CopyingChainingOutput.collect(OperatorChain.java:692) at org.apache.flink.streaming.runtime.tasks.OperatorChain$CopyingChainingOutput.collect(OperatorChain.java:672) at org.apache.flink.streaming.api.operators.CountingOutput.collect(CountingOutput.java:52) at org.apache.flink.streaming.api.operators.CountingOutput.collect(CountingOutput.java:30) at org.apache.flink.streaming.api.operators.TimestampedCollector.collect(TimestampedCollector.java:53) at org.apache.flink.table.runtime.operators.aggregate.GroupAggFunction.processElement(GroupAggFunction.java:205) at org.apache.flink.table.runtime.operators.aggregate.GroupAggFunction.processElement(GroupAggFunction.java:43) at org.apache.flink.streaming.api.operators.KeyedProcessOperator.processElement(KeyedProcessOperator.java:85) at org.apache.flink.streaming.runtime.tasks.OneInputStreamTask$StreamTaskNetworkOutput.emitRecord(OneInputStreamTask.java:161) at org.apache.flink.streaming.runtime.io.StreamTaskNetworkInput.processElement(StreamTaskNetworkInput.java:178) at org.apache.flink.streaming.runtime.io.StreamTaskNetworkInput.emitNext(StreamTaskNetworkInput.java:153) at org.apache.flink.streaming.runtime.io.StreamOneInputProcessor.processInput(StreamOneInputProcessor.java:67) at org.apache.flink.streaming.runtime.tasks.StreamTask.processInput(StreamTask.java:345) at org.apache.flink.streaming.runtime.tasks.StreamTask$$Lambda$910/1247863497.runDefaultAction(Unknown Source) at org.apache.flink.streaming.runtime.tasks.mailbox.MailboxProcessor.runMailboxStep(MailboxProcessor.java:191) at org.apache.flink.streaming.runtime.tasks.mailbox.MailboxProcessor.runMailboxLoop(MailboxProcessor.java:181) at org.apache.flink.streaming.runtime.tasks.StreamTask.runMailboxLoop(StreamTask.java:558) at org.apache.flink.streaming.runtime.tasks.StreamTask.invoke(StreamTask.java:530) at org.apache.flink.runtime.taskmanager.Task.doRun(Task.java:721) at org.apache.flink.runtime.taskmanager.Task.run(Task.java:546) at java.lang.Thread.run(Thread.java:745)
Re:Re: 请问在flinksql中如何使用聚合函数 LISTAGG
好的 , 原来是bug , 感谢回答 在 2020-08-12 21:32:40,"Benchao Li" 写道: >看起来是一个已知bug[1],已经修复,但是还没有发布。 > >[1] https://issues.apache.org/jira/browse/FLINK-18862 > >chenxuying 于2020年8月12日周三 下午9:25写道: > >> 版本: >> flinksql 1.11.0 >> 需求: >> 需要实现多行聚合成一行功能 >> 代码如下: >> environmentSettings = >> EnvironmentSettings.new_instance().in_streaming_mode().build() >> t_env = StreamTableEnvironment.create(environment_settings = >> environmentSettings) >> t_env.get_config().get_configuration().set_string("python.fn-execution.memory.managed", >> 'true') >> >> >> a_df = pd.DataFrame({"id":["1","1","1"],"uuid":["1","2","3"]}) >> a_table = t_env.from_pandas(a_df, >> DataTypes.ROW([DataTypes.FIELD("id", >> DataTypes.STRING()), >> DataTypes.FIELD("uuid", >> DataTypes.STRING())])) >> t_env.create_temporary_view("table_a",a_table) >> >> >> b_df = pd.DataFrame({"val":["3","4","5","6"],"uuid":["1","2","3","4"]}) >> table_b = t_env.from_pandas(b_df , >> DataTypes.ROW([DataTypes.FIELD("val", >> DataTypes.STRING()), >> DataTypes.FIELD("uuid", >> DataTypes.STRING())])) >> t_env.create_temporary_view("table_b",table_b) >> >> >> t_env.sql_update(""" >> CREATE TABLE mySink ( >> >> b varchar , >> c varchar >> ) WITH ( >> 'connector' = 'print' >> ) >> """) >> >> >> t_env.sql_update(""" >> insert into mySink >> select t1.id ,LISTAGG(t2.val , ',') >> from table_a t1 left join table_b t2 on t1.uuid = t2.uuid >> group by t1.id >> """) >> t_env.execute("tutorial_job") >> >> >> 报错: >> Caused by: java.lang.ClassCastException: >> org.apache.flink.table.data.binary.BinaryRawValueData cannot be cast to >> org.apache.flink.table.data.StringData at >> org.apache.flink.table.data.GenericRowData.getString(GenericRowData.java:169) >> at >> org.apache.flink.table.data.JoinedRowData.getString(JoinedRowData.java:139) >> at org.apache.flink.table.data.RowData.get(RowData.java:273) at >> org.apache.flink.table.runtime.typeutils.RowDataSerializer.copyRowData(RowDataSerializer.java:156) >> at >> org.apache.flink.table.runtime.typeutils.RowDataSerializer.copy(RowDataSerializer.java:123) >> at >> org.apache.flink.table.runtime.typeutils.RowDataSerializer.copy(RowDataSerializer.java:50) >> at >> org.apache.flink.streaming.runtime.tasks.OperatorChain$CopyingChainingOutput.pushToOperator(OperatorChain.java:715) >> at >> org.apache.flink.streaming.runtime.tasks.OperatorChain$CopyingChainingOutput.collect(OperatorChain.java:692) >> at >> org.apache.flink.streaming.runtime.tasks.OperatorChain$CopyingChainingOutput.collect(OperatorChain.java:672) >> at >> org.apache.flink.streaming.api.operators.CountingOutput.collect(CountingOutput.java:52) >> at >> org.apache.flink.streaming.api.operators.CountingOutput.collect(CountingOutput.java:30) >> at >> org.apache.flink.streaming.api.operators.TimestampedCollector.collect(TimestampedCollector.java:53) >> at >> org.apache.flink.table.runtime.operators.aggregate.GroupAggFunction.processElement(GroupAggFunction.java:205) >> at >> org.apache.flink.table.runtime.operators.aggregate.GroupAggFunction.processElement(GroupAggFunction.java:43) >> at >> org.apache.flink.streaming.api.operators.KeyedProcessOperator.processElement(KeyedProcessOperator.java:85) >> at >> org.apache.flink.streaming.runtime.tasks.OneInputStreamTask$StreamTaskNetworkOutput.emitRecord(OneInputStreamTask.java:161) >> at >> org.apache.flink.streaming.runtime.io.StreamTaskNetworkInput.processElement(StreamTaskNetworkInput.java:178) >> at >> org.apache.flink.streaming.runtime.io.StreamTaskNetworkInput.emitNext(StreamTaskNetworkInput.java:153) >> at >> org.apache.flink.streaming.runtime.io.StreamOneInputProcessor.processInput(StreamOneInputProcessor.java:67) >> at >> org.apache.flink.streaming.runtime.tasks.StreamTask.processInput(StreamTask.java:345) >> at >> org.apache.flink.streaming.runtime.tasks.StreamTask$$Lambda$910/1247863497.runDefaultAction(Unknown >> Source) at >> org.apache.flink.streaming.runtime.tasks.mailbox.MailboxProcessor.runMailboxStep(MailboxProcessor.java:191) >> at >> org.apache.flink.streaming.runtime.tasks.mailbox.MailboxProcessor.runMailboxLoop(MailboxProcessor.java:181) >> at >> org.apache.flink.streaming.runtime.tasks.StreamTask.runMailboxLoop(StreamTask.java:558) >> at >> org.apache.flink.streaming.runtime.tasks.StreamTask.invoke(StreamTask.java:530) >> at org.apache.flink.runtime.taskmanager.Task.doRun(Task.java:721) at >> org.apache.flink.runtime.taskmanager.Task.run(Task.java:546) at >> java.lang.Thread.run(Thread.java:745) >> >> >> >> > >-- > >Best, >Benchao Li
使用flinksql时 jdbc connector参数不起作用
环境是flink1.11.2+idea sql: CREATE TABLE sourceTable ( platform STRING ,game_id bigint ) WITH ( ... ); CREATE TABLE sinktable ( platform STRING ,game_id bigint ) WITH ( 'connector' = 'jdbc', 'url' = '', 'table-name' = '', 'driver' = 'com.mysql.jdbc.Driver', 'username' = '', 'password' = '', 'sink.buffer-flush.max-rows' = '2', 'sink.buffer-flush.interval' = '30s' ); insert into sinktable select platform,game_id from sourceTable; 官方文档[1]中 , 说到 sink.buffer-flush.max-rows和sink.buffer-flush.interval 这两个属性可以设置成 '0' 来禁用他 , 不过我试了下是不行 如果设置如下 sink.buffer-flush.max-rows = '0' 'sink.buffer-flush.interval' = '60s' 导致每接收一条数据就插入数据库 如果设置如下 sink.buffer-flush.max-rows = '10' 'sink.buffer-flush.interval' = '0' 导致无法插入数据库 [1]:https://ci.apache.org/projects/flink/flink-docs-release-1.11/dev/table/connectors/jdbc.html#connector-options
Re:Re: 使用flinksql时 jdbc connector参数不起作用
好的, 明白 在 2020-09-17 20:29:09,"Jark Wu" 写道: >> sink.buffer-flush.max-rows = '0' 导致每接收一条数据就插入数据库 > >这个应该是个 bug,我建了个 issue:https://issues.apache.org/jira/browse/FLINK-19280 > >Best, >Jark > >On Thu, 17 Sep 2020 at 18:15, chenxuying wrote: > >> 环境是flink1.11.2+idea >> sql: >> CREATE TABLE sourceTable ( >> platform STRING >> ,game_id bigint >> ) WITH ( >> ... >> ); >> CREATE TABLE sinktable ( >> platform STRING >> ,game_id bigint >> ) WITH ( >> 'connector' = 'jdbc', >> 'url' = '', >> 'table-name' = '', >> 'driver' = 'com.mysql.jdbc.Driver', >> 'username' = '', >> 'password' = '', >> 'sink.buffer-flush.max-rows' = '2', >> 'sink.buffer-flush.interval' = '30s' >> ); >> insert into sinktable select platform,game_id from sourceTable; >> >> >> 官方文档[1]中 , 说到 sink.buffer-flush.max-rows和sink.buffer-flush.interval >> 这两个属性可以设置成 '0' 来禁用他 , 不过我试了下是不行 >> 如果设置如下 >>sink.buffer-flush.max-rows = '0' >>'sink.buffer-flush.interval' = '60s' >> 导致每接收一条数据就插入数据库 >> 如果设置如下 >>sink.buffer-flush.max-rows = '10' >>'sink.buffer-flush.interval' = '0' >> 导致无法插入数据库 >> >> >> [1]: >> https://ci.apache.org/projects/flink/flink-docs-release-1.11/dev/table/connectors/jdbc.html#connector-options >> >>
flinksql接收到字段值格式为2020-09-23T20:58:24+08:00,如何转成TIMESTAMP
flinksql 版本是1.11.2 source接收到字段是字符串类型的时间 CREATE TABLE sourceTable ( `time` STRING ) WITH( ... ); sink如下 CREATE TABLE sinktable ( `time1` STRING, `time` TIMESTAMP(3) ) WITH ( 'connector' = 'print' ); insert语句,不知道怎么正确修改TO_TIMESTAMP默认的格式 insert into sinktable select `time`,TO_TIMESTAMP(`time`,'-MM-ddTHH:mm:ss+08:00') from sourceTable 报错说是format错误 Caused by: java.lang.IllegalArgumentException: Unknown pattern letter: T at java.time.format.DateTimeFormatterBuilder.parsePattern(DateTimeFormatterBuilder.java:1663) at java.time.format.DateTimeFormatterBuilder.appendPattern(DateTimeFormatterBuilder.java:1572) at java.time.format.DateTimeFormatter.ofPattern(DateTimeFormatter.java:534)
flink使用在docker环境中部署出现的两个问题
根据官网[1]使用docker部署flink,session cluster模式 环境win10+docker+flink1.11.2 cmd命令 docker run ^ -d^ --rm ^ --name=jobmanager ^ --hostname=jobmanager ^ --network flink-network ^ --env FLINK_PROPERTIES="jobmanager.rpc.address: jobmanager" ^ -p 28081:8081 ^ flink:1.11.2-scala_2.11 jobmanager docker run ^ -d^ --rm ^ --name=taskmanager ^ --hostname=taskmanager ^ --network flink-network ^ --env FLINK_PROPERTIES="jobmanager.rpc.address: jobmanager" ^ flink:1.11.2-scala_2.11 taskmanager 问题一: 在webui查看任务输出stdout提示找不到输出文件 java.util.concurrent.CompletionException: org.apache.flink.util.FlinkException: The file STDOUT does not exist on the TaskExecutor. at org.apache.flink.runtime.taskexecutor.TaskExecutor.lambda$requestFileUploadByFilePath$25(TaskExecutor.java:1742) ~[flink-dist_2.11-1.11.2.jar:1.11.2] at java.util.concurrent.CompletableFuture$AsyncSupply.run(CompletableFuture.java:1604) ~[?:1.8.0_265] at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149) ~[?:1.8.0_265] at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624) ~[?:1.8.0_265] at java.lang.Thread.run(Thread.java:748) ~[?:1.8.0_265] Caused by: org.apache.flink.util.FlinkException: The file STDOUT does not exist on the TaskExecutor. ... 5 more 2020-09-27 09:04:33,370 ERROR org.apache.flink.runtime.rest.handler.taskmanager.TaskManagerStdoutFileHandler [] - Unhandled exception. org.apache.flink.util.FlinkException: The file STDOUT does not exist on the TaskExecutor. at org.apache.flink.runtime.taskexecutor.TaskExecutor.lambda$requestFileUploadByFilePath$25(TaskExecutor.java:1742) ~[flink-dist_2.11-1.11.2.jar:1.11.2] at java.util.concurrent.CompletableFuture$AsyncSupply.run(CompletableFuture.java:1604) ~[?:1.8.0_265] at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149) ~[?:1.8.0_265] at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624) ~[?:1.8.0_265] at java.lang.Thread.run(Thread.java:748) ~[?:1.8.0_265] 问题二: mount的src的配置文件要多份吗 因为我把env替换成mount,如下 docker run ^ -d^ --rm ^ --name=jobmanager ^ --network flink-network ^ --mount type=bind,src=D:/cxy/soft/flink-1.11.2/conf,target=/opt/flink/conf ^ -p 28081:8081 ^ flink:1.11.2-scala_2.11 jobmanager docker run ^ -d^ --rm ^ --name=taskmanager ^ --network flink-network ^ --mount type=bind,src=D:/cxy/soft/flink-1.11.2/conf,target=/opt/flink/conf ^ flink:1.11.2-scala_2.11 taskmanager 结果发现webui上的可用Task Managers为0 每次执行命令的时候都会把mount配置的src下flink-conf.yaml中的jobmanager.rpc.address替换成了新的容器ip 我猜应该是这个原因导致启动taskmanager的时候jobmanager.rpc.address替换成了taskmanager的ip.所以没有Task可用 想问下大佬们,是我哪一步出现问题了吗 [1]:https://ci.apache.org/projects/flink/flink-docs-release-1.11/ops/deployment/docker.html
Re:Re: flink使用在docker环境中部署出现的两个问题
请问一下第一个问题您说的修改启动命令,有例子吗,或者什么文档可以参考 在 2020-09-28 12:50:25,"Yang Wang" 写道: >第一个问题,因为默认情况下JM/TM进程的STDOUT已经输出到console,所以是没有办公通过webui查看STDOUT输出的 >可以通过docker logs来查看,当然你也可以修改启动命令,把STDOUT重定向到具体的文件 > >第二个问题,在JobManager和TaskManager的docker-entrypoint.sh[1]里面会修改flink-conf.yaml的, >所以你mount进去会被修改掉 > >[1]. >https://github.com/apache/flink-docker/blob/dev-1.11/docker-entrypoint.sh > > >Best, >Yang > >chenxuying 于2020年9月27日周日 下午7:56写道: > >> 根据官网[1]使用docker部署flink,session cluster模式 >> 环境win10+docker+flink1.11.2 >> cmd命令 >> docker run ^ >> -d^ >> --rm ^ >> --name=jobmanager ^ >> --hostname=jobmanager ^ >> --network flink-network ^ >> --env FLINK_PROPERTIES="jobmanager.rpc.address: jobmanager" ^ >> -p 28081:8081 ^ >> flink:1.11.2-scala_2.11 jobmanager >> docker run ^ >> -d^ >> --rm ^ >> --name=taskmanager ^ >> --hostname=taskmanager ^ >> --network flink-network ^ >> --env FLINK_PROPERTIES="jobmanager.rpc.address: jobmanager" ^ >> flink:1.11.2-scala_2.11 taskmanager >> >> >> 问题一: >> 在webui查看任务输出stdout提示找不到输出文件 >> java.util.concurrent.CompletionException: >> org.apache.flink.util.FlinkException: The file STDOUT does not exist on the >> TaskExecutor. >> at >> org.apache.flink.runtime.taskexecutor.TaskExecutor.lambda$requestFileUploadByFilePath$25(TaskExecutor.java:1742) >> ~[flink-dist_2.11-1.11.2.jar:1.11.2] >> at >> java.util.concurrent.CompletableFuture$AsyncSupply.run(CompletableFuture.java:1604) >> ~[?:1.8.0_265] >> at >> java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149) >> ~[?:1.8.0_265] >> at >> java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624) >> ~[?:1.8.0_265] >> at java.lang.Thread.run(Thread.java:748) ~[?:1.8.0_265] >> Caused by: org.apache.flink.util.FlinkException: The file STDOUT does not >> exist on the TaskExecutor. >> ... 5 more >> 2020-09-27 09:04:33,370 ERROR >> org.apache.flink.runtime.rest.handler.taskmanager.TaskManagerStdoutFileHandler >> [] - Unhandled exception. >> org.apache.flink.util.FlinkException: The file STDOUT does not exist on >> the TaskExecutor. >> at >> org.apache.flink.runtime.taskexecutor.TaskExecutor.lambda$requestFileUploadByFilePath$25(TaskExecutor.java:1742) >> ~[flink-dist_2.11-1.11.2.jar:1.11.2] >> at >> java.util.concurrent.CompletableFuture$AsyncSupply.run(CompletableFuture.java:1604) >> ~[?:1.8.0_265] >> at >> java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149) >> ~[?:1.8.0_265] >> at >> java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624) >> ~[?:1.8.0_265] >> at java.lang.Thread.run(Thread.java:748) ~[?:1.8.0_265] >> >> >> >> >> 问题二: >> mount的src的配置文件要多份吗 >> 因为我把env替换成mount,如下 >> docker run ^ >> -d^ >> --rm ^ >> --name=jobmanager ^ >> --network flink-network ^ >> --mount type=bind,src=D:/cxy/soft/flink-1.11.2/conf,target=/opt/flink/conf >> ^ >> -p 28081:8081 ^ >> flink:1.11.2-scala_2.11 jobmanager >> >> >> docker run ^ >> -d^ >> --rm ^ >> --name=taskmanager ^ >> --network flink-network ^ >> --mount type=bind,src=D:/cxy/soft/flink-1.11.2/conf,target=/opt/flink/conf >> ^ >> flink:1.11.2-scala_2.11 taskmanager >> >> >> 结果发现webui上的可用Task Managers为0 >> 每次执行命令的时候都会把mount配置的src下flink-conf.yaml中的jobmanager.rpc.address替换成了新的容器ip >> >> 我猜应该是这个原因导致启动taskmanager的时候jobmanager.rpc.address替换成了taskmanager的ip.所以没有Task可用 >> 想问下大佬们,是我哪一步出现问题了吗 >> >> >> [1]: >> https://ci.apache.org/projects/flink/flink-docs-release-1.11/ops/deployment/docker.html
flink1.11.2基于官网在k8s上部署是正常的,但是加了volume配置之后报错Read-only file system
我在使用k8s部署的时候也是按照官网的方式[1],是正常使用的, 然后后面加了volume配置 { ... "spec": { ... "template": { ... "spec": { "volumes": [ ... { "name": "libs-volume", "hostPath": { "path": "/data/volumes/flink/jobmanager/cxylib", "type": "" } }, ... ], "containers": [ { ... "volumeMounts": [ { "name": "flink-config-volume", "mountPath": "/opt/flink/conf" }, ... ], ... } ], ... } }, ... }, ... } 然后启动jobmanager报错 Starting Job Manager sed: couldn't open temporary file /opt/flink/conf/sedz0NYKX: Read-only file system sed: couldn't open temporary file /opt/flink/conf/sede6R0BY: Read-only file system /docker-entrypoint.sh: 72: /docker-entrypoint.sh: cannot create /opt/flink/conf/flink-conf.yaml: Permission denied /docker-entrypoint.sh: 91: /docker-entrypoint.sh: cannot create /opt/flink/conf/flink-conf.yaml.tmp: Read-only file system Starting standalonesession as a console application on host flink-jobmanager-66fb98869d-w7plb. SLF4J: Failed to load class "org.slf4j.impl.StaticLoggerBinder". SLF4J: Defaulting to no-operation (NOP) logger implementation SLF4J: See http://www.slf4j.org/codes.html#StaticLoggerBinder for further details. Sep 28, 2020 7:11:14 AM org.apache.hadoop.util.NativeCodeLoader WARNING: Unable to load native-hadoop library for your platform... using builtin-java classes where applicable [1]:https://ci.apache.org/projects/flink/flink-docs-release-1.11/ops/deployment/kubernetes.html#session-cluster-resource-definitions
flinksql注册udtf使用ROW类型做为输出输出时出错
版本: pyflink==1.0 apache-flink==1.11.2 代码如下: env = StreamExecutionEnvironment.get_execution_environment() env.set_parallelism(1) t_env = StreamTableEnvironment.create(env) t_env.get_config().get_configuration().set_string("python.fn-execution.memory.managed", 'true') class SplitStr(TableFunction): def eval(self, data): for row in data: yield row[0], row[1] splitStr = udtf( SplitStr(), DataTypes.ARRAY( DataTypes.ROW( [ DataTypes.FIELD("name", DataTypes.STRING()), DataTypes.FIELD("id", DataTypes.STRING()) ] ) ), DataTypes.ROW( [ DataTypes.FIELD("name", DataTypes.STRING()), DataTypes.FIELD("id", DataTypes.STRING()) ] ) ) t_env.register_function("splitStr", splitStr) t_env.sql_update(""" CREATE TABLE mySource ( id varchar, data array> ) WITH ( 'connector' = 'kafka', 'topic' = 'mytesttopic', 'properties.bootstrap.servers' = '172.17.0.2:9092', 'properties.group.id' = 'flink-test-cxy', 'scan.startup.mode' = 'latest-offset', 'format' = 'json' ) """) t_env.sql_update(""" CREATE TABLE mysqlsink ( id varchar ,name varchar ,age varchar ) with ( 'connector' = 'print' ) """) t_env.sql_update("insert into mysqlsink select id,name,age from mySource ,LATERAL TABLE(splitStr(data)) as T(name, age)") t_env.execute("test") 最终报错 TypeError: Invalid result_type: result_type should be DataType but contains RowField(name, VARCHAR) 报错的地方是 File "C:\Users\xuyin\AppData\Local\Programs\Python\Python37\lib\site-packages\pyflink\table\udf.py", line 264, in __init__ def __init__(self, func, input_types, result_types, deterministic=None, name=None): super(UserDefinedTableFunctionWrapper, self).__init__( func, input_types, deterministic, name) if not isinstance(result_types, collections.Iterable): result_types = [result_types] for result_type in result_types: if not isinstance(result_type, DataType): raise TypeError( "Invalid result_type: result_type should be DataType but contains {}".format( result_type)) self._result_types = result_types self._judtf_placeholder = None 断点中可以看到result_types是对应着ROW里面的FIELD数组,所以报错了,这个是bug吗 另外的,假如我在 上面在创建udtf的时候,如果这样写 splitStr = udtf(SplitStr(), DataTypes.STRING(), [DataTypes.BIGINT(), DataTypes.BIGINT()]) 却可以正常运行,但是显然类型跟我实际运行的不对应
Re:flinksql注册udtf使用ROW类型做为输出输出时出错
上面最后说的splitStr = udtf(SplitStr(), DataTypes.STRING(), [DataTypes.BIGINT(), DataTypes.BIGINT()])需要改成这个地方splitStr = udtf(SplitStr(), DataTypes.STRING(), [DataTypes.STRING(), DataTypes.STRING()])udtf的第三个参数, 好像只要是能跟sink的字段类型对应就能运行, 但是第二个参数并不能跟source字段对应却能运行就感觉有点奇怪 在 2020-09-30 19:07:06,"chenxuying" 写道: >版本: >pyflink==1.0 >apache-flink==1.11.2 >代码如下: >env = StreamExecutionEnvironment.get_execution_environment() >env.set_parallelism(1) >t_env = StreamTableEnvironment.create(env) >t_env.get_config().get_configuration().set_string("python.fn-execution.memory.managed", > 'true') > > >class SplitStr(TableFunction): >def eval(self, data): >for row in data: >yield row[0], row[1] >splitStr = udtf( >SplitStr(), >DataTypes.ARRAY( >DataTypes.ROW( >[ >DataTypes.FIELD("name", DataTypes.STRING()), >DataTypes.FIELD("id", DataTypes.STRING()) >] >) >), >DataTypes.ROW( >[ >DataTypes.FIELD("name", DataTypes.STRING()), >DataTypes.FIELD("id", DataTypes.STRING()) >] >) >) >t_env.register_function("splitStr", splitStr) > > >t_env.sql_update(""" >CREATE TABLE mySource ( > >id varchar, >data array> >) WITH ( >'connector' = 'kafka', >'topic' = 'mytesttopic', >'properties.bootstrap.servers' = '172.17.0.2:9092', >'properties.group.id' = 'flink-test-cxy', >'scan.startup.mode' = 'latest-offset', >'format' = 'json' >) >""") >t_env.sql_update(""" >CREATE TABLE mysqlsink ( >id varchar >,name varchar >,age varchar >) >with ( >'connector' = 'print' >) >""") >t_env.sql_update("insert into mysqlsink select id,name,age from mySource >,LATERAL TABLE(splitStr(data)) as T(name, age)") >t_env.execute("test") > > >最终报错 >TypeError: Invalid result_type: result_type should be DataType but contains >RowField(name, VARCHAR) >报错的地方是 >File >"C:\Users\xuyin\AppData\Local\Programs\Python\Python37\lib\site-packages\pyflink\table\udf.py", > line 264, in __init__ > > >def __init__(self, func, input_types, result_types, deterministic=None, >name=None): >super(UserDefinedTableFunctionWrapper, self).__init__( >func, input_types, deterministic, name) > > >if not isinstance(result_types, collections.Iterable): >result_types = [result_types] > > >for result_type in result_types: >if not isinstance(result_type, DataType): >raise TypeError( >"Invalid result_type: result_type should be DataType but contains {}".format( >result_type)) > > >self._result_types = result_types >self._judtf_placeholder = None > > >断点中可以看到result_types是对应着ROW里面的FIELD数组,所以报错了,这个是bug吗 > > >另外的,假如我在 >上面在创建udtf的时候,如果这样写 >splitStr = udtf(SplitStr(), DataTypes.STRING(), [DataTypes.BIGINT(), >DataTypes.BIGINT()]) >却可以正常运行,但是显然类型跟我实际运行的不对应
flink1.11.2 在k8s上部署,如何启动history server
flink1.11.2 在k8s上部署,如何启动history server 之前1.10的yaml里面可以加命令,但是1.11的yaml是通过docker-entrypoint.sh 好像没发现这个入口脚本没有对应的history server参数
flink1.11.2webui界面中的任务详情页面获取不到任务的received和sent数据详情
集群是local Standalone模式,任务可以正常的运行, sink是print,能在Stdout看到数据输出, 但是在任务详情页面没有Bytes Received, Records Received, Bytes Sent , Records Sent等实时数据 都是0
Re:Re: Re: flink1.11加载外部jar包进行UDF注册
1使用系统类加载器的时候,本身作业包开放给外部UDF jar包实现的接口会报ClassNotFound异常 2线程上下文类加载器是什么 不太明白这两点,可以写个代码例子看看吗 在 2020-10-15 19:47:20,"amen...@163.com" 写道: >追加问题,在使用线程上下文类加载器的时候,数据会重复发送三条,这是因为添加pipeline.classpaths的缘故吗? >那这种设置env的方式有可能还会造成其他什么问题? > >best, >amenhub > >发件人: amen...@163.com >发送时间: 2020-10-15 19:22 >收件人: user-zh >主题: Re: Re: flink1.11加载外部jar包进行UDF注册 >非常感谢您的回复! > >对于我来说,这是一个非常好的办法,但是我有一个疑问:类加载器只能使用系统类加载器进行加载吗? >因为我在尝试使用系统类加载器的时候,本身作业包开放给外部UDF >jar包实现的接口会报ClassNotFound异常,而将类加载器指向主类(这种方式的话这里应该是使用默认的线程上下文加载器),则可避免这个问题。 > >期待您的回复,谢谢~ > >best, >amenhub >发件人: cxydeve...@163.com >发送时间: 2020-10-15 17:46 >收件人: user-zh >主题: Re: flink1.11加载外部jar包进行UDF注册 >我们用方法是通过反射设置env的配置,增加pipeline.classpaths >具体代码如下 >public static void main(final String[] args) throws Exception { >StreamExecutionEnvironment env = >StreamExecutionEnvironment.getExecutionEnvironment(); >EnvironmentSettings settings = >EnvironmentSettings.newInstance().useBlinkPlanner().inStreamingMode().build(); >StreamTableEnvironment tableEnvironment = >StreamTableEnvironment.create(env, settings); >//String path = "file:///D:/cxy/idea_workspace/...xxx.jar"; >String path = "https://...xxx.jar";; >loadJar(new URL(path)); >Field configuration = >StreamExecutionEnvironment.class.getDeclaredField("configuration"); >configuration.setAccessible(true); >Configuration o = (Configuration)configuration.get(env); >Field confData = Configuration.class.getDeclaredField("confData"); >confData.setAccessible(true); >Map temp = (Map)confData.get(o); >List jarList = new ArrayList<>(); >jarList.add(path); >temp.put("pipeline.classpaths",jarList); >tableEnvironment.executeSql("CREATE FUNCTION CxyTestReturnSelf AS >'flinksql.function.udf.CxyTestReturnSelf'"); >tableEnvironment.executeSql("CREATE TABLE sourceTable (\n" + >" f_sequence INT,\n" + >" f_random INT,\n" + >" f_random_str STRING,\n" + >" ts AS localtimestamp,\n" + >" WATERMARK FOR ts AS ts\n" + >") WITH (\n" + >" 'connector' = 'datagen',\n" + >" 'rows-per-second'='5',\n" + >"\n" + >" 'fields.f_sequence.kind'='sequence',\n" + >" 'fields.f_sequence.start'='1',\n" + >" 'fields.f_sequence.end'='1000',\n" + >"\n" + >" 'fields.f_random.min'='1',\n" + >" 'fields.f_random.max'='1000',\n" + >"\n" + >" 'fields.f_random_str.length'='10'\n" + >")"); >tableEnvironment.executeSql("CREATE TABLE sinktable (\n" + >"f_random_str STRING" + >") WITH (\n" + >"'connector' = 'print'\n" + >")"); >tableEnvironment.executeSql( >"insert into sinktable " + >"select CxyTestReturnSelf(f_random_str) " + >"from sourceTable"); >} >//动态加载Jar >public static void loadJar(URL jarUrl) { >//从URLClassLoader类加载器中获取类的addURL方法 >Method method = null; >try { >method = URLClassLoader.class.getDeclaredMethod("addURL", >URL.class); >} catch (NoSuchMethodException | SecurityException e1) { >e1.printStackTrace(); >} >// 获取方法的访问权限 >boolean accessible = method.isAccessible(); >try { >//修改访问权限为可写 >if (accessible == false) { >method.setAccessible(true); >} >// 获取系统类加载器 >URLClassLoader classLoader = (URLClassLoader) >ClassLoader.getSystemClassLoader(); >//jar路径加入到系统url路径里 >method.invoke(classLoader, jarUrl); >} catch (Exception e) { >e.printStackTrace(); >} finally { >method.setAccessible(accessible); >} >} >-- >Sent from: http://apache-flink.147419.n8.nabble.com/
请问flink1.11版本如何设置checkpoint的默认保存个数
我看官方文档[1]应该是设置state.checkpoints.num-retained , 默认是1, 但是设置了没有效果, 官方说默认是1, 但是我发现好像是10 , 同时我也设置了其他的属性,比如 execution.checkpointing.externalized-checkpoint-retention: RETAIN_ON_CANCELLATION 是可行,所以我的设置应该没有什么问题 [1]:https://ci.apache.org/projects/flink/flink-docs-stable/dev/stream/state/checkpointing.html#state-checkpoints-num-retained
flinksql当一个job的kafka连接信息错误时,会导致整个session集群无法正常发布任务
环境: flinksql 1.12.2 k8s session模式 描述: 当kafka 端口错误,过一段时间会有如下报错: 2021-04-25 16:49:50 org.apache.kafka.common.errors.TimeoutException: Timeout of 6ms expired before the position for partition filebeat_json_install_log-3 could be determined 当kafka ip错误,过一段时间会有如下报错: 2021-04-25 20:12:53 org.apache.flink.runtime.JobException: Recovery is suppressed by NoRestartBackoffTimeStrategy at org.apache.flink.runtime.executiongraph.failover.flip1.ExecutionFailureHandler.handleFailure(ExecutionFailureHandler.java:118) at org.apache.flink.runtime.executiongraph.failover.flip1.ExecutionFailureHandler.getFailureHandlingResult(ExecutionFailureHandler.java:80) at org.apache.flink.runtime.scheduler.DefaultScheduler.handleTaskFailure(DefaultScheduler.java:233) at org.apache.flink.runtime.scheduler.DefaultScheduler.maybeHandleTaskFailure(DefaultScheduler.java:224) at org.apache.flink.runtime.scheduler.DefaultScheduler.updateTaskExecutionStateInternal(DefaultScheduler.java:215) at org.apache.flink.runtime.scheduler.SchedulerBase.updateTaskExecutionState(SchedulerBase.java:669) at org.apache.flink.runtime.scheduler.SchedulerNG.updateTaskExecutionState(SchedulerNG.java:89) at org.apache.flink.runtime.jobmaster.JobMaster.updateTaskExecutionState(JobMaster.java:447) at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method) at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62) at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43) at java.lang.reflect.Method.invoke(Method.java:498) at org.apache.flink.runtime.rpc.akka.AkkaRpcActor.handleRpcInvocation(AkkaRpcActor.java:305) at org.apache.flink.runtime.rpc.akka.AkkaRpcActor.handleRpcMessage(AkkaRpcActor.java:212) at org.apache.flink.runtime.rpc.akka.FencedAkkaRpcActor.handleRpcMessage(FencedAkkaRpcActor.java:77) at org.apache.flink.runtime.rpc.akka.AkkaRpcActor.handleMessage(AkkaRpcActor.java:158) at akka.japi.pf.UnitCaseStatement.apply(CaseStatements.scala:26) at akka.japi.pf.UnitCaseStatement.apply(CaseStatements.scala:21) at scala.PartialFunction$class.applyOrElse(PartialFunction.scala:123) at akka.japi.pf.UnitCaseStatement.applyOrElse(CaseStatements.scala:21) at scala.PartialFunction$OrElse.applyOrElse(PartialFunction.scala:170) at scala.PartialFunction$OrElse.applyOrElse(PartialFunction.scala:171) at scala.PartialFunction$OrElse.applyOrElse(PartialFunction.scala:171) at akka.actor.Actor$class.aroundReceive(Actor.scala:517) at akka.actor.AbstractActor.aroundReceive(AbstractActor.scala:225) at akka.actor.ActorCell.receiveMessage(ActorCell.scala:592) at akka.actor.ActorCell.invoke(ActorCell.scala:561) at akka.dispatch.Mailbox.processMailbox(Mailbox.scala:258) at akka.dispatch.Mailbox.run(Mailbox.scala:225) at akka.dispatch.Mailbox.exec(Mailbox.scala:235) at akka.dispatch.forkjoin.ForkJoinTask.doExec(ForkJoinTask.java:260) at akka.dispatch.forkjoin.ForkJoinPool$WorkQueue.runTask(ForkJoinPool.java:1339) at akka.dispatch.forkjoin.ForkJoinPool.runWorker(ForkJoinPool.java:1979) at akka.dispatch.forkjoin.ForkJoinWorkerThread.run(ForkJoinWorkerThread.java:107) Caused by: org.apache.kafka.common.errors.TimeoutException: Timeout expired while fetching topic metadata 然后对任务执行停止取消操作,会得到如下错误 2021-04-25 08:53:41,115 INFO org.apache.flink.runtime.executiongraph.ExecutionGraph [] - Job v2_ods_device_action_log (fcc451b8a521398b10e5b86153141fbf) switched from state CANCELLING to CANCELED. 2021-04-25 08:53:41,115 INFO org.apache.flink.runtime.checkpoint.CheckpointCoordinator[] - Stopping checkpoint coordinator for job fcc451b8a521398b10e5b86153141fbf. 2021-04-25 08:53:41,115 INFO org.apache.flink.runtime.checkpoint.StandaloneCompletedCheckpointStore [] - Shutting down 2021-04-25 08:53:41,115 INFO org.apache.flink.runtime.checkpoint.CompletedCheckpoint [] - Checkpoint with ID 1 at 'oss://tanwan-datahub/test/flinksql/checkpoints/fcc451b8a521398b10e5b86153141fbf/chk-1' not discarded. 2021-04-25 08:53:41,115 INFO org.apache.flink.runtime.checkpoint.CompletedCheckpoint [] - Checkpoint with ID 2 at 'oss://tanwan-datahub/test/flinksql/checkpoints/fcc451b8a521398b10e5b86153141fbf/chk-2' not discarded. 2021-04-25 08:53:41,116 INFO org.apache.flink.runtime.checkpoint.CompletedCheckpoint [] - Checkpoint with ID 3 at 'oss://tanwan-datahub/test/flinksql/checkpoints/fcc451b8a521398b10e5b86153141fbf/chk-3' not discarded. 2021-04-25 08:53:41,137 INFO org.apache.flink.runtime.dispatcher.StandaloneDispatcher [] - Job fcc451b8a521398b10e5b86153141fbf reached globally terminal state CANCELED. 2021-04-25 08:53:41,148 INFO org.apache.flink.runtime.jobmaster.JobMaster [] - Stopping the JobMaster for job v2_ods_device_action_log(fcc451b8a521398b10e5b86153141fbf). 2021-04-25 08:53:41,151 INFO org.apache.flink.runtime.jobmaster.slotpool.SlotPoolImpl [] - Suspending SlotPool. 2021-04-25 08:53:41,151 INFO
来自chenxuying的邮件
flink启动任务的方式
请问下目前flink的启动方式有哪些 1 通过命令行来执行 flink run -C file:///usr/local/soft/flink/function-0.1.jar -c cn.xxx.flink.table.sql.Job /usr/local/soft/flink/flink-1.0-SNAPSHOT.jar 2通过自带的webui页面上传jar , submit jar 3 通过代码 createRemoteEnvironment 目前主要使用的是,通过代码请求rest api http请求来实现上传jar,跑任务, 但是目前遇到的问题是,通过rest api 无法实现命令行那样提供其他的jar包
Re:Re: flink启动任务的方式
这个是可以 , 不过我们的需求不允许打FatJar 在 2020-04-21 15:27:48,"Arnold Zai" 写道: >打个FatJar > >chenxuying 于2020年4月21日周二 下午2:47写道: > >> 请问下目前flink的启动方式有哪些 >> 1 通过命令行来执行 >> flink run -C file:///usr/local/soft/flink/function-0.1.jar -c >> cn.xxx.flink.table.sql.Job /usr/local/soft/flink/flink-1.0-SNAPSHOT.jar >> 2通过自带的webui页面上传jar , submit jar >> 3 通过代码 createRemoteEnvironment >> >> 目前主要使用的是,通过代码请求rest api http请求来实现上传jar,跑任务, 但是目前遇到的问题是,通过rest api >> 无法实现命令行那样提供其他的jar包 >> >> >> >>
Re:Re: Re: flink启动任务的方式
您说的jarFiles是以什么样的方式提交任务 然后我试了一下plugin,好像并不可以,重启flink cluster也不行 , 也不知是不是我的方式不对 我的目录结构是 xxx/flink/plugins/ folder1/ udf.jar 另外说一下,如果我把udf.jar放到 /flink/lib下,重启是可以的,不过这不是我想要的方式,不知道您是否理解,因为我想要的我随时可以写个udf.jar,随时可以用,不要重启flink cluster 在 2020-04-21 17:46:00,"Arnold Zai" 写道: >jarFiles参数不是个参数列表么,多传几个。 > >或把依赖提前部署到${FLINK_HOME}/plugins里 > >chenxuying 于2020年4月21日周二 下午3:36写道: > >> 这个是可以 , 不过我们的需求不允许打FatJar >> >> >> >> >> >> >> >> >> >> >> >> >> >> >> >> >> >> 在 2020-04-21 15:27:48,"Arnold Zai" 写道: >> >打个FatJar >> > >> >chenxuying 于2020年4月21日周二 下午2:47写道: >> > >> >> 请问下目前flink的启动方式有哪些 >> >> 1 通过命令行来执行 >> >> flink run -C file:///usr/local/soft/flink/function-0.1.jar -c >> >> cn.xxx.flink.table.sql.Job /usr/local/soft/flink/flink-1.0-SNAPSHOT.jar >> >> 2通过自带的webui页面上传jar , submit jar >> >> 3 通过代码 createRemoteEnvironment >> >> >> >> 目前主要使用的是,通过代码请求rest api http请求来实现上传jar,跑任务, 但是目前遇到的问题是,通过rest api >> >> 无法实现命令行那样提供其他的jar包 >> >> >> >> >> >> >> >> >>
Re:Re: Re: FlinkSQL 是否支持类似临时中间表的概念
你好,请问下,my_parse是个udf吧 然后有没有什么操作可以使用udtf解析出多个字段 , 这些字段直接就是source表的字段 , 然后选出时间字段定义 watermark , 类似如下 CREATE TABLE sourceTable( request_uri STRING, (column_1,column_2,heart_time) as udtf_parse(request_uri) )with(..); 哈哈,不知道有没有这样的语法 在 2020-06-24 12:24:46,"Jark Wu" 写道: >你可以在 DDL 中直接用计算列去从 request_uri 里获得 heart_time 哈,然后在这个计算列上定义 watermark 即可。 >例如: > >CREATE TABLE sourceTable ( > request_uri STRING, > heart_time AS my_parse(request_uri), > WATERMARK FOR heart_time AS heart_time - INTERVAL '1' SECOND >) WITH ( ... ); > >虽然这会导致重复解析两遍。 > > >Best, >Jark > >On Wed, 24 Jun 2020 at 12:09, Weixubin <18925434...@163.com> wrote: > >> >> >> >> 感谢你提供了子查询的思路,不过经过试验后有点可惜,这似乎还是满足不了我们的需求。 >> >> >> 我们的场景是从阿里云SLS读取消息。每条消息有一个字段是request_uri。 >> 第一个数据处理过程就是将 request_uri 解析为多个属性(多列),存成一行,作为一条记录。 >> 第二个数据处理过程就是将每条记录里的 声明heart_time为事件时间属性并使用5秒延迟水印策略,进行开窗聚合处理。 >> >> >> //如果应用到子查询的话,Flink是不支持这样做的。WATERMARK FOR 水印声明只能在DDL里应用。如下: >> select >> ..ts as TO_TIMESTAMP(heart_time,'-MM-ddHH:mm:ss') , >> WATERMARK FOR ts AS ts - INTERVAL '1' SECOND >> from sourceTable , LATERAL TABLE(ParseUriRow(request_uri)) as T( )….· >> >> >> //如果应用到source,则一开始并不知道heart_time 的值 >> CREATE TABLE sourceTable ( >> request_uri STRING >> ..ts as TO_TIMESTAMP(heart_time,'-MM-ddHH:mm:ss') , >> WATERMARK FOR ts AS ts - INTERVAL '1' SECOND >> ) WITH ( ... ); >> >> >> 只能等待Flink 1.11 尝试是否可以用View作为中间临时表,并对View进行 WATERMARK水印声明 >> Thanks >> Bin >> >> 在 2020-06-23 15:28:50,"Leonard Xu" 写道: >> >Hi >> >我的意思是你如果中间结果表如果要输出,那你就一个sink写到中间结果表(什么表根据你的需要),一个sink写到你的最终结果表,在这两个sink之前的`select >> * from sourceTable , LATERAL TABLE(ParseUriRow(request_uri)) as T( )….` >> 的这段sql是可以复用的,就和 VIEW的作用类似。 >> > >> >如果你不需要中间结果表,只是要最终的结果表,那你写个嵌套的sql就行了,里层是`select * from sourceTable , >> LATERAL TABLE(ParseUriRow(request_uri)) as T( )….·,外层是 group by, >> 插入最终的结果表就能满足需求了吧。 >> > >> >祝好, >> >Leonard Xu >> > >> > >> >> 在 2020年6月23日,15:21,Weixubin <18925434...@163.com> 写道: >> >> >> >> >> >> >> >> >> >> Hi, >> >> 关于这句 “把 ` select * from sourceTable , LATERAL >> TABLE(ParseUriRow(request_uri)) as T( )….` 这段 sql insert 到中间结果表 和 >> group后再写入最终结果表就可以了” >> >> 我不太明白所说的中间结果表是什么意思, 我所理解为在数据库创建多一张middleSinkTable表,作为中间结果表。请问这样理解是否有误? >> 可否简单举个例子。 >> >> Thanks, >> >> Bin >> >> >> >> >> >> >> >> >> >> >> >> >> >> >> >> >> >> >> >> >> >> 在 2020-06-23 11:57:28,"Leonard Xu" 写道: >> >>> Hi, >> >>> 是的,这个是在 FLINK 1.11 的版本里支持的,当前 FLINK 1.11代码已经冻结,正在测试中,最近应该快发布了,所以master >> 分支上的版本号为1.12-SNAPSHOT >> >>> ,等1.11版本发布了就可以看到对应的文档。 >> >>> >> >>> 回到这个问题,flink sql(blink planner) 支持 multiple-sinks 的, 在 1.10 >> 版本也可以一个作业搞定。 把 ` select * from sourceTable , LATERAL >> TABLE(ParseUriRow(request_uri)) as T( )….` 这段 sql insert 到中间结果表 和 >> group后再写入最终结果表就可以了。效果和使用 VIEW 应该类似,因为 planner 会做分段优化。 >> >>> 另外建议使用1.10.1版本,1.10.1在1.10的基础上修复了不少bug,也可以等1.11发布了试用下1.11。 >> >>> >> >>> >> >>> 祝好, >> >>> Leonard Xu >> > >>
flink REST API是否支持-C参数
目前使用的是flink 1.10.0 背景: REST API有一个提交job的接口 接口 /jars/:jarid/run 参数entryClass,programArgs,parallelism,jobId,allowNonRestoredState,savepointPath 如果使用命令行方式提交job flink run -C file:///usr/local/soft/flink/my-function-0.1.jar -c cn.xuying.flink.table.sql.ParserSqlJob /usr/local/soft/flink/flink-1.0-SNAPSHOT.jar 可以看到命令行方式支持-C提供另外的jar包,flink会加载到classpath 问题: 发现目前的restapi并没有提供想命令行一样的-C参数的功能 , 所以想知道这个功能将来是否会增加