Error when use Kyuubi flink engine wirte data from kafka to mysql #2294

lichaozg · 2022-04-07T11:54:41Z

lichaozg
Apr 7, 2022

CREATE TABLE source_600_kafka ( `age` INT,
`name` string ) WITH ( 'connector' = 'kafka',
'sink.parallelism' = '10',
'properties.bootstrap.servers' = '***********',
'topic' = 'dio_test3',
'scan.startup.mode' = 'latest-offset',
'properties.group.id' = 'dio_test3Source',
'format' = 'json' );
CREATE TABLE sink_600_jdbc ( `id` INT,
`name` string,
PRIMARY KEY (id) NOT ENFORCED ) WITH ( 'connector' = 'jdbc',
'password' = '123456',
'url' = 'jdbc:mysql://*********/flink_test',
'table-name' = 'flink_ddl3',
'username' = 'root' );
INSERT INTO sink_600_jdbc SELECT age,name FROM source_600_kafka;

报的错误

 jdbc:hive2://bigdata-server-03:10009/> INSERT INTO sink_600_jdbc SELECT age,name FROM source_600_kafka;
19:34:50.277 INFO org.apache.kyuubi.operation.ExecuteStatement: Processing lctest's query[6faee6c2-0b7e-4a7c-8769-572545af212c]: INITIALIZED_STATE -> PENDING_STATE, statement: INSERT INTO sink_600_jdbc SELECT age,name FROM source_600_kafka
19:34:50.279 INFO org.apache.kyuubi.operation.ExecuteStatement: Processing lctest's query[6faee6c2-0b7e-4a7c-8769-572545af212c]: PENDING_STATE -> RUNNING_STATE, statement: INSERT INTO sink_600_jdbc SELECT age,name FROM source_600_kafka
19:34:50.340 INFO org.apache.kyuubi.operation.ExecuteStatement: Query[6faee6c2-0b7e-4a7c-8769-572545af212c] in ERROR_STATE
19:34:50.340 INFO org.apache.kyuubi.operation.ExecuteStatement: Processing lctest's query[6faee6c2-0b7e-4a7c-8769-572545af212c]: RUNNING_STATE -> ERROR_STATE, statement: INSERT INTO sink_600_jdbc SELECT age,name FROM source_600_kafka, time taken: 0.061 seconds
Error: Error operating EXECUTE_STATEMENT: org.apache.flink.table.client.gateway.SqlExecutionException: Could not execute SQL statement.
        at org.apache.flink.table.client.gateway.local.LocalExecutor.executeOperation(LocalExecutor.java:211)
        at org.apache.kyuubi.engine.flink.operation.ExecuteStatement.runOperation(ExecuteStatement.scala:165)
        at org.apache.kyuubi.engine.flink.operation.ExecuteStatement.org$apache$kyuubi$engine$flink$operation$ExecuteStatement$$executeStatement(ExecuteStatement.scala:113)
        at org.apache.kyuubi.engine.flink.operation.ExecuteStatement$$anon$1.run(ExecuteStatement.scala:78)
        at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511)
        at java.util.concurrent.FutureTask.run(FutureTask.java:266)
        at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
        at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
        at java.lang.Thread.run(Thread.java:748)
Caused by: java.lang.IllegalArgumentException: No handler for method [public abstract java.util.Set org.apache.flink.table.planner.plan.metadata.FlinkMetadata$UpsertKeys.getUpsertKeys()] applied to argument of type [interface org.apache.calcite.rel.RelNode]; we recommend you create a catch-all (RelNode) handler
        at org.apache.calcite.rel.metadata.ReflectiveRelMetadataProvider$Space.find(ReflectiveRelMetadataProvider.java:320)
        at org.apache.calcite.rel.metadata.JaninoRelMetadataProvider.load3(JaninoRelMetadataProvider.java:339)
        at org.apache.calcite.rel.metadata.JaninoRelMetadataProvider.lambda$static$0(JaninoRelMetadataProvider.java:109)
        at org.apache.flink.calcite.shaded.com.google.common.cache.CacheLoader$FunctionToCacheLoader.load(CacheLoader.java:165)
        at org.apache.flink.calcite.shaded.com.google.common.cache.LocalCache$LoadingValueReference.loadFuture(LocalCache.java:3529)
        at org.apache.flink.calcite.shaded.com.google.common.cache.LocalCache$Segment.loadSync(LocalCache.java:2278)
        at org.apache.flink.calcite.shaded.com.google.common.cache.LocalCache$Segment.lockedGetOrLoad(LocalCache.java:2155)
        at org.apache.flink.calcite.shaded.com.google.common.cache.LocalCache$Segment.get(LocalCache.java:2045)
        at org.apache.flink.calcite.shaded.com.google.common.cache.LocalCache.get(LocalCache.java:3951)
        at org.apache.flink.calcite.shaded.com.google.common.cache.LocalCache.getOrLoad(LocalCache.java:3974)
        at org.apache.flink.calcite.shaded.com.google.common.cache.LocalCache$LocalLoadingCache.get(LocalCache.java:4958)
        at org.apache.calcite.rel.metadata.JaninoRelMetadataProvider.create(JaninoRelMetadataProvider.java:469)
        at org.apache.calcite.rel.metadata.JaninoRelMetadataProvider.revise(JaninoRelMetadataProvider.java:481)
        at org.apache.calcite.rel.metadata.RelMetadataQueryBase.revise(RelMetadataQueryBase.java:95)
        at org.apache.flink.table.planner.plan.metadata.FlinkRelMetadataQuery.getUpsertKeys(FlinkRelMetadataQuery.java:287)
        at org.apache.flink.table.planner.plan.optimize.program.FlinkChangelogModeInferenceProgram$SatisfyUpdateKindTraitVisitor.inferSinkRequiredTraits(FlinkChangelogModeInferenceProgram.scala:781)
        at org.apache.flink.table.planner.plan.optimize.program.FlinkChangelogModeInferenceProgram$SatisfyUpdateKindTraitVisitor.visit(FlinkChangelogModeInferenceProgram.scala:436)
        at org.apache.flink.table.planner.plan.optimize.program.FlinkChangelogModeInferenceProgram.$anonfun$optimize$1(FlinkChangelogModeInferenceProgram.scala:80)
        at scala.collection.TraversableLike.$anonfun$flatMap$1(TraversableLike.scala:240)
        at scala.collection.immutable.List.foreach(List.scala:388)
        at scala.collection.TraversableLike.flatMap(TraversableLike.scala:240)
        at scala.collection.TraversableLike.flatMap$(TraversableLike.scala:237)
        at scala.collection.immutable.List.flatMap(List.scala:351)
        at org.apache.flink.table.planner.plan.optimize.program.FlinkChangelogModeInferenceProgram.optimize(FlinkChangelogModeInferenceProgram.scala:79)
        at org.apache.flink.table.planner.plan.optimize.program.FlinkChangelogModeInferenceProgram.optimize(FlinkChangelogModeInferenceProgram.scala:46)
        at org.apache.flink.table.planner.plan.optimize.program.FlinkGroupProgram.$anonfun$optimize$2(FlinkGroupProgram.scala:63)
        at scala.collection.TraversableOnce.$anonfun$foldLeft$1(TraversableOnce.scala:156)
        at scala.collection.TraversableOnce.$anonfun$foldLeft$1$adapted(TraversableOnce.scala:156)
        at scala.collection.Iterator.foreach(Iterator.scala:937)
        at scala.collection.Iterator.foreach$(Iterator.scala:937)
        at scala.collection.AbstractIterator.foreach(Iterator.scala:1425)
        at scala.collection.IterableLike.foreach(IterableLike.scala:70)
        at scala.collection.IterableLike.foreach$(IterableLike.scala:69)
        at scala.collection.AbstractIterable.foreach(Iterable.scala:54)
        at scala.collection.TraversableOnce.foldLeft(TraversableOnce.scala:156)
        at scala.collection.TraversableOnce.foldLeft$(TraversableOnce.scala:154)
        at scala.collection.AbstractTraversable.foldLeft(Traversable.scala:104)
        at org.apache.flink.table.planner.plan.optimize.program.FlinkGroupProgram.$anonfun$optimize$1(FlinkGroupProgram.scala:60)
        at org.apache.flink.table.planner.plan.optimize.program.FlinkGroupProgram.$anonfun$optimize$1$adapted(FlinkGroupProgram.scala:55)
        at scala.collection.TraversableOnce.$anonfun$foldLeft$1(TraversableOnce.scala:156)
        at scala.collection.TraversableOnce.$anonfun$foldLeft$1$adapted(TraversableOnce.scala:156)
        at scala.collection.immutable.Range.foreach(Range.scala:155)
        at scala.collection.TraversableOnce.foldLeft(TraversableOnce.scala:156)
        at scala.collection.TraversableOnce.foldLeft$(TraversableOnce.scala:154)
        at scala.collection.AbstractTraversable.foldLeft(Traversable.scala:104)
        at org.apache.flink.table.planner.plan.optimize.program.FlinkGroupProgram.optimize(FlinkGroupProgram.scala:55)
        at org.apache.flink.table.planner.plan.optimize.program.FlinkChainedProgram.$anonfun$optimize$1(FlinkChainedProgram.scala:62)
        at scala.collection.TraversableOnce.$anonfun$foldLeft$1(TraversableOnce.scala:156)
        at scala.collection.TraversableOnce.$anonfun$foldLeft$1$adapted(TraversableOnce.scala:156)
        at scala.collection.Iterator.foreach(Iterator.scala:937)
        at scala.collection.Iterator.foreach$(Iterator.scala:937)
        at scala.collection.AbstractIterator.foreach(Iterator.scala:1425)
        at scala.collection.IterableLike.foreach(IterableLike.scala:70)
        at scala.collection.IterableLike.foreach$(IterableLike.scala:69)
        at scala.collection.AbstractIterable.foreach(Iterable.scala:54)
        at scala.collection.TraversableOnce.foldLeft(TraversableOnce.scala:156)
        at scala.collection.TraversableOnce.foldLeft$(TraversableOnce.scala:154)
        at scala.collection.AbstractTraversable.foldLeft(Traversable.scala:104)
        at org.apache.flink.table.planner.plan.optimize.program.FlinkChainedProgram.optimize(FlinkChainedProgram.scala:58)
        at org.apache.flink.table.planner.plan.optimize.StreamCommonSubGraphBasedOptimizer.optimizeTree(StreamCommonSubGraphBasedOptimizer.scala:161)
        at org.apache.flink.table.planner.plan.optimize.StreamCommonSubGraphBasedOptimizer.doOptimize(StreamCommonSubGraphBasedOptimizer.scala:79)
        at org.apache.flink.table.planner.plan.optimize.CommonSubGraphBasedOptimizer.optimize(CommonSubGraphBasedOptimizer.scala:77)
        at org.apache.flink.table.planner.delegation.PlannerBase.optimize(PlannerBase.scala:300)
        at org.apache.flink.table.planner.delegation.PlannerBase.translate(PlannerBase.scala:183)
        at org.apache.flink.table.api.internal.TableEnvironmentImpl.translate(TableEnvironmentImpl.java:1665)
        at org.apache.flink.table.api.internal.TableEnvironmentImpl.executeInternal(TableEnvironmentImpl.java:752)
        at org.apache.flink.table.api.internal.TableEnvironmentImpl.executeInternal(TableEnvironmentImpl.java:872)
        at org.apache.flink.table.client.gateway.local.LocalExecutor.lambda$executeOperation$3(LocalExecutor.java:209)
        at org.apache.flink.table.client.gateway.context.ExecutionContext.wrapClassLoader(ExecutionContext.java:88)
        at org.apache.flink.table.client.gateway.local.LocalExecutor.executeOperation(LocalExecutor.java:209)
        ... 8 more (state=,code=0)

Answered by a49a

Apr 8, 2022

Delete PRIMARY KEY (id) NOT ENFORCED in the MySQL CREATE TABLE statement and the SQL will run successfully.

View full answer

a49a · 2022-04-08T11:58:32Z

a49a
Apr 8, 2022

Delete PRIMARY KEY (id) NOT ENFORCED in the MySQL CREATE TABLE statement and the SQL will run successfully.

1 reply

lichaozg Apr 9, 2022
Author

Thank you

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Uh oh!

Error when use Kyuubi flink engine wirte data from kafka to mysql #2294

Uh oh!

{{title}}

Uh oh!

Uh oh!

{{editor}}'s edit

{{editor}}'s edit

Uh oh!

Replies: 1 comment 1 reply

Uh oh!

{{title}}

Uh oh!

Uh oh!

{{title}}

Uh oh!

Select a reply

Uh oh!

Error when use Kyuubi flink engine wirte data from kafka to mysql #2294

Uh oh!

Uh oh!

lichaozg Apr 7, 2022

Replies: 1 comment · 1 reply

Uh oh!

a49a Apr 8, 2022

Uh oh!

lichaozg Apr 9, 2022 Author

lichaozg
Apr 7, 2022

Replies: 1 comment 1 reply

a49a
Apr 8, 2022

lichaozg Apr 9, 2022
Author