开发者社区> 问答> 正文

flink-1.11使用executeSql()执行DDL语句问题

环境信息:flink-1.11.0, blink-planner, 本地ide开发测试(IDEA)

问题描述:使用executeSql()方法执行DDL语句,控制台打印如下异常信息。(flink-connector-kafka_2.11依赖已添加)

我不确定是否还有某个必要的依赖没有添加,还是有其他的地方没有考虑完整,请大佬赐教。

--------------------------------------------------------------分割线-------------------------------------------------------------

Exception in thread "main" org.apache.flink.table.api.ValidationException: Unable to create a source for reading table 'default_catalog.default_database.kafka_out'.

Table options are:

'connector'='kafka'

'format'='json'

'properties.bootstrap.servers'='localhost:9092'

'properties.group.id'='flink-1.11'

'scan.startup.mode'='group-offsets'

'topic'='flink-kafka'

at org.apache.flink.table.factories.FactoryUtil.createTableSource(FactoryUtil.java:125)

at org.apache.flink.table.planner.plan.schema.CatalogSourceTable.buildTableScan(CatalogSourceTable.scala:135)

at org.apache.flink.table.planner.plan.schema.CatalogSourceTable.toRel(CatalogSourceTable.scala:78)

at org.apache.calcite.sql2rel.SqlToRelConverter.toRel(SqlToRelConverter.java:3492)

at org.apache.calcite.sql2rel.SqlToRelConverter.convertIdentifier(SqlToRelConverter.java:2415)

at org.apache.calcite.sql2rel.SqlToRelConverter.convertFrom(SqlToRelConverter.java:2102)

at org.apache.calcite.sql2rel.SqlToRelConverter.convertFrom(SqlToRelConverter.java:2051)

at org.apache.calcite.sql2rel.SqlToRelConverter.convertSelectImpl(SqlToRelConverter.java:661)

at org.apache.calcite.sql2rel.SqlToRelConverter.convertSelect(SqlToRelConverter.java:642)

at org.apache.calcite.sql2rel.SqlToRelConverter.convertQueryRecursive(SqlToRelConverter.java:3345)

at org.apache.calcite.sql2rel.SqlToRelConverter.convertFrom(SqlToRelConverter.java:2178)

at org.apache.calcite.sql2rel.SqlToRelConverter.convertFrom(SqlToRelConverter.java:2051)

at org.apache.calcite.sql2rel.SqlToRelConverter.convertSelectImpl(SqlToRelConverter.java:661)

at org.apache.calcite.sql2rel.SqlToRelConverter.convertSelect(SqlToRelConverter.java:642)

at org.apache.calcite.sql2rel.SqlToRelConverter.convertQueryRecursive(SqlToRelConverter.java:3345)

at org.apache.calcite.sql2rel.SqlToRelConverter.convertQuery(SqlToRelConverter.java:568)

at org.apache.flink.table.planner.calcite.FlinkPlannerImpl.org$apache$flink$table$planner$calcite$FlinkPlannerImpl$$rel(FlinkPlannerImpl.scala:164)

at org.apache.flink.table.planner.calcite.FlinkPlannerImpl.rel(FlinkPlannerImpl.scala:151)

at org.apache.flink.table.planner.operations.SqlToOperationConverter.toQueryOperation(SqlToOperationConverter.java:773)

at org.apache.flink.table.planner.operations.SqlToOperationConverter.convertSqlQuery(SqlToOperationConverter.java:745)

at org.apache.flink.table.planner.operations.SqlToOperationConverter.convert(SqlToOperationConverter.java:238)

at org.apache.flink.table.planner.operations.SqlToOperationConverter.convertSqlInsert(SqlToOperationConverter.java:527)

at org.apache.flink.table.planner.operations.SqlToOperationConverter.convert(SqlToOperationConverter.java:204)

at org.apache.flink.table.planner.delegation.ParserImpl.parse(ParserImpl.java:78)

at org.apache.flink.table.api.internal.TableEnvironmentImpl.executeSql(TableEnvironmentImpl.java:678)

at example.Example.main(Example.java:77)

Caused by: org.apache.flink.table.api.ValidationException: Could not find any factories that implement 'org.apache.flink.table.factories.DeserializationFormatFactory' in the classpath.

at org.apache.flink.table.factories.FactoryUtil.discoverFactory(FactoryUtil.java:229)

at org.apache.flink.table.factories.FactoryUtil$TableFactoryHelper.discoverOptionalFormatFactory(FactoryUtil.java:538)

at org.apache.flink.table.factories.FactoryUtil$TableFactoryHelper.discoverOptionalDecodingFormat(FactoryUtil.java:426)

at org.apache.flink.table.factories.FactoryUtil$TableFactoryHelper.discoverDecodingFormat(FactoryUtil.java:413)

at org.apache.flink.streaming.connectors.kafka.table.KafkaDynamicTableFactoryBase.createDynamicTableSource(KafkaDynamicTableFactoryBase.java:73)

at org.apache.flink.table.factories.FactoryUtil.createTableSource(FactoryUtil.java:122)

... 25 more

*来自志愿者整理的flink邮件归档

展开
收起
小阿矿 2021-12-06 15:44:04 1064 0
1 条回答
写回答
取消 提交回答
  • 看起来是format找不到实现,你可以添加一下flink-json的依赖看一下。

    *来自志愿者整理的flink邮件归档

    2021-12-06 18:06:25
    赞同 展开评论 打赏
问答排行榜
最热
最新

相关电子书

更多
Flink SQL in 2020 立即下载
Flink Streaming SQL 2018 立即下载
基于 Flink SQL + Paimon 构建流式湖仓新方 立即下载