Hi
I have a question when use flink sql 1.11.0
Error:
Exception in thread "main"
org.apache.flink.table.api.NoMatchingTableFactoryException: Could not find a
suitable table factory for 'org.apache.flink.table.delegation.ExecutorFactory'
in
the classpath.
Reason: No factory implements
'org.apache.flink.table.delegation.ExecutorFactory'.
The following properties are requested:
class-name=org.apache.flink.table.planner.delegation.BlinkExecutorFactory
streaming-mode=true
The following factories have been considered:
org.apache.flink.table.module.CoreModuleFactory
org.apache.flink.table.sources.CsvBatchTableSourceFactory
org.apache.flink.table.sources.CsvAppendTableSourceFactory
org.apache.flink.table.sinks.CsvBatchTableSinkFactory
org.apache.flink.table.sinks.CsvAppendTableSinkFactory
org.apache.flink.table.catalog.GenericInMemoryCatalogFactory
org.apache.flink.table.filesystem.FileSystemTableFactory
org.apache.flink.api.java.io.jdbc.JDBCTableSourceSinkFactory
org.apache.flink.connector.jdbc.catalog.factory.JdbcCatalogFactory
org.apache.flink.connector.jdbc.table.JdbcTableSourceSinkFactory
at
org.apache.flink.table.factories.TableFactoryService.filterByFactoryClass(TableFactoryService.java:238)
at
org.apache.flink.table.factories.TableFactoryService.filter(TableFactoryService.java:185)
at
org.apache.flink.table.factories.TableFactoryService.findAllInternal(TableFactoryService.java:171)
at
org.apache.flink.table.factories.TableFactoryService.findAll(TableFactoryService.java:125)
at
org.apache.flink.table.factories.ComponentFactoryService.find(ComponentFactoryService.java:48)
at
org.apache.flink.table.api.internal.TableEnvironmentImpl.create(TableEnvironmentImpl.java:257)
at org.apache.flink.table.api.TableEnvironment.create(TableEnvironment.java:93)
at com.weifeng.demo.JoinDemo04$.main(JoinDemo04.scala:15)
at com.weifeng.demo.JoinDemo04.main(JoinDemo04.scala)
My code is:
val settings =
EnvironmentSettings.newInstance().useBlinkPlanner().inStreamingMode().build()
val tableEnv = TableEnvironment.create(settings)
val mapedStream: DataStream[util.Map[String, Any]] =
environment.readTextFile("/Users/weifeng/IdeaProjects/flinksqldemo/src/main/resources/demo.txt",
"utf-8")
.map(x => {
val splits = x.split(",", -1)
var map1: util.Map[String, Any] = new util.LinkedHashMap[String, Any]()
map1.put("mobile_id", splits(0))
map1.put("name", splits(1))
map1
})
val mapedStream1: DataStream[util.Map[String, Any]] =
environment.readTextFile("/Users/weifeng/IdeaProjects/flinksqldemo/src/main/resources/demo.txt",
"utf-8")
.map(x => {
val splits = x.split(",", -1)
var map1: util.Map[String, Any] = new util.LinkedHashMap[String, Any]()
map1.put("mobile_id", splits(0))
map1.put("name", splits(1))
map1
})
val name = "mypg"
val defaultDatabase = "bdhtest"
val username = "bdh"
val password = "flRfRO9Lovgc"
val baseUrl = "jdbc:postgresql://10.218.1.190:8433/bdhtest”
val catalog = new JdbcCatalog(name, defaultDatabase, username, password,
baseUrl)
tableEnv.registerCatalog("mypg", catalog)
tableEnv.useCatalog("mypg")
tableEnv.execute("JoinDemo04”)
Thank You
原始邮件
发件人: Caizhi Weng (Jira)<[email protected]>
收件人: dev<[email protected]>
发送时间: 2020年8月5日(周三) 14:59
主题: [jira] [Created] (FLINK-18821) Netty client retry mechanism maycause
PartitionRequestClientFactory#createPartitionRequestClient to waitinfinitely
Caizhi Weng created FLINK-18821:
-----------------------------------
Summary: Netty client retry mechanism may cause
PartitionRequestClientFactory#createPartitionRequestClient to wait infinitely
Key: FLINK-18821
URL: https://issues.apache.org/jira/browse/FLINK-18821
Project: Flink
Issue Type: Bug
Components: Runtime / Network
Affects Versions: 1.11.0
Reporter: Caizhi Weng
When running TPCDS 10T benchmark on Flink I found some of the task slots stuck.
After some investigation there seems to be a bug in
{{PartitionRequestClientFactory}}.
When a task tries to require a partition of data from its upstream task but
fails, {{PartitionRequestClientFactory#connect}} will throw
{{RemoteTransportException}} and
{{PartitionRequestClientFactory#connectWithRetries}} will throw
{{CompletionException}}. However this exception is not caught by
{{PartitionRequestClientFactory#connect}} and it will eventually fail the task.
But {{PartitionRequestClientFactory}} lives in a task manager not in a task. In
{{PartitionRequestClientFactory}} a {{ConcurrentHashMap}} named {{clients}} is
maintained for reusing {{NettyPartitionRequestClient}}. When the above
exception happens, {{clients}} is not cleaned up; When the next call to
{{PartitionRequestClientFactory#connect}} with the same connection id comes, it
will use the invalid {{CompletableFuture}} in {{clients}} and this future will
never complete, causing the task to stuck forever.
Exception stack:
{code}
2020-08-05 03:37:07,539 ERROR
org.apache.flink.runtime.io.network.netty.PartitionRequestClientFactory [] -
Failed 1 times to connect to <host-name>/<ip>:<port>
org.apache.flink.runtime.io.network.netty.exception.RemoteTransportException:
Connecting to remote task manager '<host-name>/<ip>:<port>' has failed. This
might indicate that the remote task manager has been lost.
at
org.apache.flink.runtime.io.network.netty.PartitionRequestClientFactory.connect(PartitionRequestClientFactory.java:120)
~[flink-dist_2.11-1.12-SNAPSHOT.jar:1.12-SNAPSHOT]
at
org.apache.flink.runtime.io.network.netty.PartitionRequestClientFactory.connectWithRetries(PartitionRequestClientFactory.java:99)
~[flink-dist_2.11-1.12-SNAPSHOT.jar:1.12-SNAPSHOT]
at
org.apache.flink.runtime.io.network.netty.PartitionRequestClientFactory.createPartitionRequestClient(PartitionRequestClientFactory.java:76)
~[flink-dist_2.11-1.12-SNAPSHOT.jar:1.12-SNAPSHOT]
at
org.apache.flink.runtime.io.network.netty.NettyConnectionManager.createPartitionRequestClient(NettyConnectionManager.java:67)
~[flink-dist_2.11-1.12-SNAPSHOT.jar:1.12-SNAPSHOT]
at
org.apache.flink.runtime.io.network.partition.consumer.RemoteInputChannel.requestSubpartition(RemoteInputChannel.java:146)
~[flink-dist_2.11-1.12-SNAPSHOT.jar:1.12-SNAPSHOT]
at
org.apache.flink.runtime.io.network.partition.consumer.SingleInputGate.internalRequestPartitions(SingleInputGate.java:329)
~[flink-dist_2.11-1.12-SNAPSHOT.jar:1.12-SNAPSHOT]
at
org.apache.flink.runtime.io.network.partition.consumer.SingleInputGate.requestPartitions(SingleInputGate.java:301)
~[flink-dist_2.11-1.12-SNAPSHOT.jar:1.12-SNAPSHOT]
at
org.apache.flink.runtime.taskmanager.InputGateWithMetrics.requestPartitions(InputGateWithMetrics.java:95)
~[flink-dist_2.11-1.12-SNAPSHOT.jar:1.12-SNAPSHOT]
at
org.apache.flink.streaming.runtime.tasks.StreamTask.requestPartitions(StreamTask.java:514)
~[flink-dist_2.11-1.12-SNAPSHOT.jar:1.12-SNAPSHOT]
at
org.apache.flink.streaming.runtime.tasks.StreamTask.readRecoveredChannelState(StreamTask.java:484)
~[flink-dist_2.11-1.12-SNAPSHOT.jar:1.12-SNAPSHOT]
at
org.apache.flink.streaming.runtime.tasks.StreamTask.lambda$beforeInvoke$0(StreamTask.java:475)
~[flink-dist_2.11-1.12-SNAPSHOT.jar:1.12-SNAPSHOT]
at
org.apache.flink.streaming.runtime.tasks.StreamTaskActionExecutor$1.runThrowing(StreamTaskActionExecutor.java:47)
[flink-dist_2.11-1.12-SNAPSHOT.jar:1.12-SNAPSHOT]
at
org.apache.flink.streaming.runtime.tasks.StreamTask.beforeInvoke(StreamTask.java:469)
[flink-dist_2.11-1.12-SNAPSHOT.jar:1.12-SNAPSHOT]
at
org.apache.flink.streaming.runtime.tasks.StreamTask.invoke(StreamTask.java:522)
[flink-dist_2.11-1.12-SNAPSHOT.jar:1.12-SNAPSHOT]
at org.apache.flink.runtime.taskmanager.Task.doRun(Task.java:721)
[flink-dist_2.11-1.12-SNAPSHOT.jar:1.12-SNAPSHOT]
at org.apache.flink.runtime.taskmanager.Task.run(Task.java:546)
[flink-dist_2.11-1.12-SNAPSHOT.jar:1.12-SNAPSHOT]
at java.lang.Thread.run(Thread.java:834) [?:1.8.0_102]
Caused by: java.lang.NullPointerException
at
org.apache.flink.util.Preconditions.checkNotNull(Preconditions.java:58)
~[flink-dist_2.11-1.12-SNAPSHOT.jar:1.12-SNAPSHOT]
at
org.apache.flink.runtime.io.network.netty.NettyPartitionRequestClient.<init>(NettyPartitionRequestClient.java:73)
~[flink-dist_2.11-1.12-SNAPSHOT.jar:1.12-SNAPSHOT]
at
org.apache.flink.runtime.io.network.netty.PartitionRequestClientFactory.connect(PartitionRequestClientFactory.java:114)
~[flink-dist_2.11-1.12-SNAPSHOT.jar:1.12-SNAPSHOT]
... 16 more
{code}
--
This message was sent by Atlassian Jira
(v8.3.4#803005)
