升級(jí)flink1.14.4報(bào)錯(cuò)?

Caused by: org.apache.flink.table.api.ValidationException: Unable to create a source for reading table 'default_catalog.default_database.new_buyer_trade_order2'??

CAUSED BY: 2022-03-11 16:45:04,169 INFO  org.apache.flink.yarn.cli.FlinkYarnSessionCli                [] - Dynamic Property set: metrics.reporter.influxdb.class=org.apache.flink.metrics.influxdb.InfluxdbReporter
2022-03-11 16:45:04,169 INFO  org.apache.flink.yarn.cli.FlinkYarnSessionCli                [] - Dynamic Property set: metrics.reporter.influxdb.class=org.apache.flink.metrics.influxdb.InfluxdbReporter
2022-03-11 16:45:04,170 INFO  org.apache.flink.yarn.cli.FlinkYarnSessionCli                [] - Dynamic Property set: state.backend.rocksdb.ttl.compaction.filter.enabled=true
2022-03-11 16:45:04,170 INFO  org.apache.flink.yarn.cli.FlinkYarnSessionCli                [] - Dynamic Property set: state.backend.rocksdb.ttl.compaction.filter.enabled=true
2022-03-11 16:45:04,171 INFO  org.apache.flink.yarn.cli.FlinkYarnSessionCli                [] - Dynamic Property set: metrics.reporter.influxdb.db=flink
2022-03-11 16:45:04,171 INFO  org.apache.flink.yarn.cli.FlinkYarnSessionCli                [] - Dynamic Property set: metrics.reporter.influxdb.db=flink
2022-03-11 16:45:04,171 INFO  org.apache.flink.yarn.cli.FlinkYarnSessionCli                [] - Dynamic Property set: metrics.reporter.influxdb.host=192.168.5.57
2022-03-11 16:45:04,171 INFO  org.apache.flink.yarn.cli.FlinkYarnSessionCli                [] - Dynamic Property set: metrics.reporter.influxdb.host=192.168.5.57
2022-03-11 16:45:04,171 INFO  org.apache.flink.yarn.cli.FlinkYarnSessionCli                [] - Dynamic Property set: web.timeout=120000
2022-03-11 16:45:04,171 INFO  org.apache.flink.yarn.cli.FlinkYarnSessionCli                [] - Dynamic Property set: web.timeout=120000
2022-03-11 16:45:04,171 INFO  org.apache.flink.yarn.cli.FlinkYarnSessionCli                [] - Dynamic Property set: akka.ask.timeout=120 s
2022-03-11 16:45:04,171 INFO  org.apache.flink.yarn.cli.FlinkYarnSessionCli                [] - Dynamic Property set: akka.ask.timeout=120 s
2022-03-11 16:45:04,171 INFO  org.apache.flink.yarn.cli.FlinkYarnSessionCli                [] - Dynamic Property set: env.java.opts=-verbose:gc -XX:NewRatio=3 -XX:+UseConcMarkSweepGC -XX:+PrintGCDetails -XX:+PrintGCDateStamps -XX:ParallelGCThreads=4 -Duser.timezone=Asia/Shanghai
2022-03-11 16:45:04,171 INFO  org.apache.flink.yarn.cli.FlinkYarnSessionCli                [] - Dynamic Property set: env.java.opts=-verbose:gc -XX:NewRatio=3 -XX:+UseConcMarkSweepGC -XX:+PrintGCDetails -XX:+PrintGCDateStamps -XX:ParallelGCThreads=4 -Duser.timezone=Asia/Shanghai
2022-03-11 16:45:04,171 INFO  org.apache.flink.yarn.cli.FlinkYarnSessionCli                [] - Dynamic Property set: metrics.reporter.influxdb.port=8086
2022-03-11 16:45:04,171 INFO  org.apache.flink.yarn.cli.FlinkYarnSessionCli                [] - Dynamic Property set: metrics.reporter.influxdb.port=8086
2022-03-11 16:45:04,171 INFO  org.apache.flink.yarn.cli.FlinkYarnSessionCli                [] - Dynamic Property set: akka.watch.heartbeat.interval=10 s
2022-03-11 16:45:04,171 INFO  org.apache.flink.yarn.cli.FlinkYarnSessionCli                [] - Dynamic Property set: akka.watch.heartbeat.interval=10 s
2022-03-11 16:45:04,172 INFO  org.apache.flink.yarn.cli.FlinkYarnSessionCli                [] - Dynamic Property set: slotmanager.taskmanager-timeout=600000
2022-03-11 16:45:04,172 INFO  org.apache.flink.yarn.cli.FlinkYarnSessionCli                [] - Dynamic Property set: slotmanager.taskmanager-timeout=600000
2022-03-11 16:45:04,172 INFO  org.apache.flink.yarn.cli.FlinkYarnSessionCli                [] - Dynamic Property set: containerized.heap-cutoff-min=100
2022-03-11 16:45:04,172 INFO  org.apache.flink.yarn.cli.FlinkYarnSessionCli                [] - Dynamic Property set: containerized.heap-cutoff-min=100
建表語(yǔ)句解析成功!connector:upsert-kafka
設(shè)置kafka時(shí)間撮:upsert-kafka
建表語(yǔ)句解析成功!connector:print
hive conf path:/etc/ecm/hive-conf
hive conf path:/etc/ecm/hive-conf
create iceberg catalog success!
start to run sql:CREATE TABLE new_buyer_trade_order2 (
  database VARCHAR,  `table`  VARCHAR,  type  VARCHAR,  ts   BIGINT,  xid   BIGINT,  xoffset  BIGINT,  data  VARCHAR,    `old`   VARCHAR      
) WITH (
'value.format' = 'json',
'key.format' = 'json',
'properties.bootstrap.servers' = '192.168.8.142:9092,192.168.8.141:9092,192.168.8.143:9092',
'connector' = 'upsert-kafka',
'topic' = 'new_buyer_trade_order2')
start to run sql:create table result_print(   database VARCHAR,  `table`  VARCHAR,  type  VARCHAR,  ts   BIGINT,  xid   BIGINT,  xoffset  BIGINT,  data  VARCHAR,    `old`   VARCHAR )with(     'connector' = 'print' )
start to run sql:insert into result_print select      database,  `table`,  type,  ts,  xid,  xoffset,  database,    `old` from new_buyer_trade_order2

------------------------------------------------------------
 The program finished with the following exception:

org.apache.flink.client.program.ProgramInvocationException: The main method caused an error: Unable to create a source for reading table 'default_catalog.default_database.new_buyer_trade_order2'.

Table options are:

'connector'='upsert-kafka'
'key.format'='json'
'properties.bootstrap.servers'='192.168.8.142:9092,192.168.8.141:9092,192.168.8.143:9092'
'topic'='new_buyer_trade_order2'
'value.format'='json'
	at org.apache.flink.client.program.PackagedProgram.callMainMethod(PackagedProgram.java:372)
	at org.apache.flink.client.program.PackagedProgram.invokeInteractiveModeForExecution(PackagedProgram.java:222)
	at org.apache.flink.client.ClientUtils.executeProgram(ClientUtils.java:114)
	at org.apache.flink.client.cli.CliFrontend.executeProgram(CliFrontend.java:812)
	at org.apache.flink.client.cli.CliFrontend.run(CliFrontend.java:246)
	at org.apache.flink.client.cli.CliFrontend.parseAndRun(CliFrontend.java:1054)
	at org.apache.flink.client.cli.CliFrontend.lambda$main$10(CliFrontend.java:1132)
	at java.security.AccessController.doPrivileged(Native Method)
	at javax.security.auth.Subject.doAs(Subject.java:422)
	at org.apache.hadoop.security.UserGroupInformation.doAs(UserGroupInformation.java:1836)
	at org.apache.flink.runtime.security.contexts.HadoopSecurityContext.runSecured(HadoopSecurityContext.java:41)
	at org.apache.flink.client.cli.CliFrontend.main(CliFrontend.java:1132)
Caused by: org.apache.flink.table.api.ValidationException: Unable to create a source for reading table 'default_catalog.default_database.new_buyer_trade_order2'.

Table options are:

'connector'='upsert-kafka'
'key.format'='json'
'properties.bootstrap.servers'='192.168.8.142:9092,192.168.8.141:9092,192.168.8.143:9092'
'topic'='new_buyer_trade_order2'
'value.format'='json'
	at org.apache.flink.table.factories.FactoryUtil.createTableSource(FactoryUtil.java:150)
	at org.apache.flink.table.planner.plan.schema.CatalogSourceTable.createDynamicTableSource(CatalogSourceTable.java:116)
	at org.apache.flink.table.planner.plan.schema.CatalogSourceTable.toRel(CatalogSourceTable.java:82)
	at org.apache.calcite.sql2rel.SqlToRelConverter.toRel(SqlToRelConverter.java:3585)
	at org.apache.calcite.sql2rel.SqlToRelConverter.convertIdentifier(SqlToRelConverter.java:2507)
	at org.apache.calcite.sql2rel.SqlToRelConverter.convertFrom(SqlToRelConverter.java:2144)
	at org.apache.calcite.sql2rel.SqlToRelConverter.convertFrom(SqlToRelConverter.java:2093)
	at org.apache.calcite.sql2rel.SqlToRelConverter.convertFrom(SqlToRelConverter.java:2050)
	at org.apache.calcite.sql2rel.SqlToRelConverter.convertSelectImpl(SqlToRelConverter.java:663)
	at org.apache.calcite.sql2rel.SqlToRelConverter.convertSelect(SqlToRelConverter.java:644)
	at org.apache.calcite.sql2rel.SqlToRelConverter.convertQueryRecursive(SqlToRelConverter.java:3438)
	at org.apache.calcite.sql2rel.SqlToRelConverter.convertQuery(SqlToRelConverter.java:570)
	at org.apache.flink.table.planner.calcite.FlinkPlannerImpl.org$apache$flink$table$planner$calcite$FlinkPlannerImpl$$rel(FlinkPlannerImpl.scala:177)
	at org.apache.flink.table.planner.calcite.FlinkPlannerImpl.rel(FlinkPlannerImpl.scala:169)
	at org.apache.flink.table.planner.operations.SqlToOperationConverter.toQueryOperation(SqlToOperationConverter.java:1057)
	at org.apache.flink.table.planner.operations.SqlToOperationConverter.convertSqlQuery(SqlToOperationConverter.java:1026)
	at org.apache.flink.table.planner.operations.SqlToOperationConverter.convert(SqlToOperationConverter.java:301)
	at org.apache.flink.table.planner.operations.SqlToOperationConverter.convertSqlInsert(SqlToOperationConverter.java:639)
	at org.apache.flink.table.planner.operations.SqlToOperationConverter.convert(SqlToOperationConverter.java:290)
	at org.apache.flink.table.planner.delegation.ParserImpl.parse(ParserImpl.java:101)
	at org.apache.flink.table.api.internal.TableEnvironmentImpl.sqlUpdate(TableEnvironmentImpl.java:834)
	at com.gegejia.flink.FlinkJobBoot.main(FlinkJobBoot.java:95)
	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.client.program.PackagedProgram.callMainMethod(PackagedProgram.java:355)
	... 11 more
Caused by: org.apache.flink.table.api.ValidationException: 'upsert-kafka' tables require to define a PRIMARY KEY constraint. The PRIMARY KEY specifies which columns should be read from or write to the Kafka message key. The PRIMARY KEY also defines records in the 'upsert-kafka' table should update or delete on which keys.
	at org.apache.flink.streaming.connectors.kafka.table.UpsertKafkaDynamicTableFactory.validatePKConstraints(UpsertKafkaDynamicTableFactory.java:261)
	at org.apache.flink.streaming.connectors.kafka.table.UpsertKafkaDynamicTableFactory.validateSource(UpsertKafkaDynamicTableFactory.java:219)
	at org.apache.flink.streaming.connectors.kafka.table.UpsertKafkaDynamicTableFactory.createDynamicTableSource(UpsertKafkaDynamicTableFactory.java:119)
	at org.apache.flink.table.factories.FactoryUtil.createTableSource(FactoryUtil.java:147)
	... 37 more

?source表未加主鍵導(dǎo)致，注釋放開(kāi)，提交成功

文章來(lái)源地址http://www.zghlxwxcb.cn/news/detail-610572.html

到了這里，關(guān)于[flink1.14.4]Unable to create a source for reading table ‘default_catalog.default_database.new_buyer的文章就介紹完了。如果您還想了解更多內(nèi)容，請(qǐng)?jiān)谟疑辖撬阉鱐OY模板網(wǎng)以前的文章或繼續(xù)瀏覽下面的相關(guān)文章，希望大家以后多多支持TOY模板網(wǎng)！