问题一:flink TableEnvironment.sqlUpdate不支持update 多表关联更新吗
诸同仁好,flink TableEnvironment.sqlUpdate是不是不支持update 多表关联更新? 如下代码: bbTableEnv.sqlUpdate("update order_tb a, min_max_usertime_tb b set a.mark=-2 " + " where a.mac=b.mac and extract(epoch from a.usertime)/7200 = b.user_2hour " + " and a.usertime > b.min_usertime and a.usertime < b.max_usertime"); 报错如下: Exception in thread "main" org.apache.flink.table.api.SqlParserException: SQL parse failed. Encountered "," at line 1, column 17. Was expecting: "SET" ... at org.apache.flink.table.planner.calcite.FlinkPlannerImpl.parse(FlinkPlannerImpl.scala:96) at org.apache.flink.table.planner.delegation.PlannerBase.parse(PlannerBase.scala:127) at org.apache.flink.table.api.internal.TableEnvironmentImpl.sqlUpdate(TableEnvironmentImpl.java:335) at com.sir.idle.IdleAnalysis.runBlinkBatch(IdleAnalysis.java:101) at com.sir.BatchMain.main(BatchMain.java:17) Caused by: org.apache.calcite.sql.parser.SqlParseException: Encountered "," at line 1, column 17. Was expecting: "SET" ... at org.apache.flink.sql.parser.impl.FlinkSqlParserImpl.convertException(FlinkSqlParserImpl.java:368) at org.apache.flink.sql.parser.impl.FlinkSqlParserImpl.normalizeException(FlinkSqlParserImpl.java:167) at org.apache.calcite.sql.parser.SqlParser.handleException(SqlParser.java:147) at org.apache.calcite.sql.parser.SqlParser.parseQuery(SqlParser.java:162) at org.apache.calcite.sql.parser.SqlParser.parseStmt(SqlParser.java:187) at org.apache.flink.table.planner.calcite.FlinkPlannerImpl.parse(FlinkPlannerImpl.scala:92) ... 4 more Caused by: org.apache.flink.sql.parser.impl.ParseException: Encountered "," at line 1, column 17. Was expecting: "SET" ... at org.apache.flink.sql.parser.impl.FlinkSqlParserImpl.generateParseException(FlinkSqlParserImpl.java:33107) at org.apache.flink.sql.parser.impl.FlinkSqlParserImpl.jj_consume_token(FlinkSqlParserImpl.java:32921) at org.apache.flink.sql.parser.impl.FlinkSqlParserImpl.SqlUpdate(FlinkSqlParserImpl.java:8227) at org.apache.flink.sql.parser.impl.FlinkSqlParserImpl.SqlStmt(FlinkSqlParserImpl.java:3646) at org.apache.flink.sql.parser.impl.FlinkSqlParserImpl.SqlStmtEof(FlinkSqlParserImpl.java:3669) at org.apache.flink.sql.parser.impl.FlinkSqlParserImpl.parseSqlStmtEof(FlinkSqlParserImpl.java:214) at org.apache.calcite.sql.parser.SqlParser.parseQuery(SqlParser.java:160) ... 6 more*来自志愿者整理的flink邮件归档
参考回答:
SQL 也不能这样吧
关于本问题的更多回答可点击原文查看:https://developer.aliyun.com/ask/359636?spm=a2c6h.13262185.0.0.7c83253fTqut4U
问题二:编译Flink1.11的flink-runtime-web失败怎么办?
hi,all
我在编译Flink1.11,由于每次到flink-runtime-web都失败,于是我cd flink-runtime-web进行单独编译,发现
Cannot resolve org.apache.flink:flink-test-utils-junit:1.11-SNAPSHOT,
Cannot resolve org.apache.flink:flink-test-utils_2.11:1.11-SNAPSHOT
依赖一直无法下载下来。请问有好的解决方法吗?
感谢你们在百忙之中看到我的邮件!*来自志愿者整理的flink邮件归档
参考回答:
为什么还会依赖 -SNAPSHOT 的jar。不是release 的 的版本吗?
关于本问题的更多回答可点击原文查看:https://developer.aliyun.com/ask/359637?spm=a2c6h.13262185.0.0.7c83253fTqut4U
问题三:Flink checkpoint 速度慢问题怎么办?
谢谢回复 看了数次checkpoint慢的情况,发现大多是async阶段耗时,如果是这样,那这应该是那个时刻网络原因导致的慢吧? 但是我还是觉得跟磁盘有一定关系 *来自志愿者整理的flink邮件归档
参考回答:
能通过日志或监控判断是 checkpoint 时 snapshot 的 sync 阶段慢,还是 async 阶段慢,还是上传到 HDFS
时间长或是其他阶段的瓶颈吗?
几十 KB 的状态慢很可能是某个步骤出故障卡住了。
关于本问题的更多回答可点击原文查看:https://developer.aliyun.com/ask/359107?spm=a2c6h.13262185.0.0.4cf552d9esI2u2
问题四:Flink 1.12 ApplicationMode运行在阿里云托管Kubernetes报错怎么办?
使用Flink1.12 Application Mode在阿里云托管Kubernetes ACK启动发现一些报错,同样的报错在自建Kubernetes集群中未发现。
但是观察taskmanager容器有正常启动,后续任务也可正常执行,针对该报错需如何处理?是不兼容阿里云ACK集群么?
启动命令:
./bin/flink run-application \
--target kubernetes-application \
-Dkubernetes.cluster-id=demo \
-Dkubernetes.container.image=xx.xx.xx/xx/xxx:2.0.12 \
local:///opt/flink/usrlib/my-flink-job.jar
日志:
2021-03-01 04:52:06,518 INFO org.apache.flink.client.deployment.application.executors.EmbeddedExecutor [] - Job 6eb4027586e7137b20ecc8c3ce624417 is submitted.
2021-03-01 04:52:06,518 INFO org.apache.flink.client.deployment.application.executors.EmbeddedExecutor [] - Submitting Job with JobId=6eb4027586e7137b20ecc8c3ce624417.
2021-03-01 04:52:08,303 INFO org.apache.flink.kubernetes.KubernetesResourceManagerDriver [] - Recovered 0 pods from previous attempts, current attempt id is 1.
2021-03-01 04:52:08,303 INFO org.apache.flink.runtime.resourcemanager.active.ActiveResourceManager [] - Recovered 0 workers from previous attempt.
2021-03-01 04:52:08,306 INFO org.apache.flink.runtime.resourcemanager.active.ActiveResourceManager [] - ResourceManager akka.tcp://flink@demo.default:6123/user/rpc/resourcemanager_0 was granted leadership with fencing token 00000000000000000000000000000000
2021-03-01 04:52:08,310 INFO org.apache.flink.runtime.resourcemanager.slotmanager.SlotManagerImpl [] - Starting the SlotManager.
2021-03-01 04:52:08,596 WARN org.apache.flink.runtime.jobmaster.MiniDispatcherRestEndpoint [] - Unhandled exception
java.io.IOException: Connection reset by peer
at sun.nio.ch.FileDispatcherImpl.read0(Native Method) ~[?:1.8.0_275]
at sun.nio.ch.SocketDispatcher.read(SocketDispatcher.java:39) ~[?:1.8.0_275]
at sun.nio.ch.IOUtil.readIntoNativeBuffer(IOUtil.java:223) ~[?:1.8.0_275]
at sun.nio.ch.IOUtil.read(IOUtil.java:192) ~[?:1.8.0_275]
at sun.nio.ch.SocketChannelImpl.read(SocketChannelImpl.java:379) ~[?:1.8.0_275]
at org.apache.flink.shaded.netty4.io.netty.buffer.PooledByteBuf.setBytes(PooledByteBuf.java:253) ~[flink-dist_2.12-1.12.0.jar:1.12.0]
at org.apache.flink.shaded.netty4.io.netty.buffer.AbstractByteBuf.writeBytes(AbstractByteBuf.java:1133) ~[flink-dist_2.12-1.12.0.jar:1.12.0]
at org.apache.flink.shaded.netty4.io.netty.channel.socket.nio.NioSocketChannel.doReadBytes(NioSocketChannel.java:350) ~[flink-dist_2.12-1.12.0.jar:1.12.0]
at org.apache.flink.shaded.netty4.io.netty.channel.nio.AbstractNioByteChannel$NioByteUnsafe.read(AbstractNioByteChannel.java:148) [flink-dist_2.12-1.12.0.jar:1.12.0]
at org.apache.flink.shaded.netty4.io.netty.channel.nio.NioEventLoop.processSelectedKey(NioEventLoop.java:714) [flink-dist_2.12-1.12.0.jar:1.12.0]
at org.apache.flink.shaded.netty4.io.netty.channel.nio.NioEventLoop.processSelectedKeysOptimized(NioEventLoop.java:650) [flink-dist_2.12-1.12.0.jar:1.12.0]
at org.apache.flink.shaded.netty4.io.netty.channel.nio.NioEventLoop.processSelectedKeys(NioEventLoop.java:576) [flink-dist_2.12-1.12.0.jar:1.12.0]
at org.apache.flink.shaded.netty4.io.netty.channel.nio.NioEventLoop.run(NioEventLoop.java:493) [flink-dist_2.12-1.12.0.jar:1.12.0]
at org.apache.flink.shaded.netty4.io.netty.util.concurrent.SingleThreadEventExecutor$4.run(SingleThreadEventExecutor.java:989) [flink-dist_2.12-1.12.0.jar:1.12.0]
at org.apache.flink.shaded.netty4.io.netty.util.internal.ThreadExecutorMap$2.run(ThreadExecutorMap.java:74) [flink-dist_2.12-1.12.0.jar:1.12.0]
at java.lang.Thread.run(Thread.java:748) [?:1.8.0_275]
2021-03-01 04:52:08,596 WARN org.apache.flink.runtime.jobmaster.MiniDispatcherRestEndpoint [] - Unhandled exception
java.io.IOException: Connection reset by peer
at sun.nio.ch.FileDispatcherImpl.read0(Native Method) ~[?:1.8.0_275]
at sun.nio.ch.SocketDispatcher.read(SocketDispatcher.java:39) ~[?:1.8.0_275]
at sun.nio.ch.IOUtil.readIntoNativeBuffer(IOUtil.java:223) ~[?:1.8.0_275]
at sun.nio.ch.IOUtil.read(IOUtil.java:192) ~[?:1.8.0_275]
at sun.nio.ch.SocketChannelImpl.read(SocketChannelImpl.java:379) ~[?:1.8.0_275]
at org.apache.flink.shaded.netty4.io.netty.buffer.PooledByteBuf.setBytes(PooledByteBuf.java:253) ~[flink-dist_2.12-1.12.0.jar:1.12.0]
at org.apache.flink.shaded.netty4.io.netty.buffer.AbstractByteBuf.writeBytes(AbstractByteBuf.java:1133) ~[flink-dist_2.12-1.12.0.jar:1.12.0]
at org.apache.flink.shaded.netty4.io.netty.channel.socket.nio.NioSocketChannel.doReadBytes(NioSocketChannel.java:350) ~[flink-dist_2.12-1.12.0.jar:1.12.0]
at org.apache.flink.shaded.netty4.io.netty.channel.nio.AbstractNioByteChannel$NioByteUnsafe.read(AbstractNioByteChannel.java:148) [flink-dist_2.12-1.12.0.jar:1.12.0]
at org.apache.flink.shaded.netty4.io.netty.channel.nio.NioEventLoop.processSelectedKey(NioEventLoop.java:714) [flink-dist_2.12-1.12.0.jar:1.12.0]
at org.apache.flink.shaded.netty4.io.netty.channel.nio.NioEventLoop.processSelectedKeysOptimized(NioEventLoop.java:650) [flink-dist_2.12-1.12.0.jar:1.12.0]
at org.apache.flink.shaded.netty4.io.netty.channel.nio.NioEventLoop.processSelectedKeys(NioEventLoop.java:576) [flink-dist_2.12-1.12.0.jar:1.12.0]
at org.apache.flink.shaded.netty4.io.netty.channel.nio.NioEventLoop.run(NioEventLoop.java:493) [flink-dist_2.12-1.12.0.jar:1.12.0]
at org.apache.flink.shaded.netty4.io.netty.util.concurrent.SingleThreadEventExecutor$4.run(SingleThreadEventExecutor.java:989) [flink-dist_2.12-1.12.0.jar:1.12.0]
at org.apache.flink.shaded.netty4.io.netty.util.internal.ThreadExecutorMap$2.run(ThreadExecutorMap.java:74) [flink-dist_2.12-1.12.0.jar:1.12.0]
at java.lang.Thread.run(Thread.java:748) [?:1.8.0_275]
后续不停日志滚动刷新exception
参考回答:
这个其实原因是阿里云的LoadBalancer探活机制不停的给Flink的rest endpoint发送RST导致的 目前有一个ticket来跟进这个问题[1],但还没有修复
短时间内你可以通过log4j的配置将org.apache.flink.runtime.jobmaster.MiniDispatcherRestEndpoint 这个package的log level设置为WARN来暂时避免
[1]. https://issues.apache.org/jira/browse/FLINK-18129
关于本问题的更多回答可点击原文查看:https://developer.aliyun.com/ask/359112
问题五:关于rebalance和forward的性能问题提问?
如题,如果我确认某2个task我期望是2个task,即不会chain到一起。这种情况下使用foward和rebalannce有哪种固定会性能更好吗。哪种一般性能更好呢?
参考回答:
你好,
不 chain 的话基本取决于上游是否有数据倾斜吧。
有数据倾斜的话 Rebalance 可以让下游算子均衡一点,性能会好一些;没有数据倾斜的话,Forward 是一对一发,Rebalance 是 n 对 m 发,后者网络开销相对会大一些,Forward 的性能可能会稍微好一点点,但是我理解这个差异对作业性能的影响可能微乎其微。
关于本问题的更多回答可点击原文查看:https://developer.aliyun.com/ask/359123