备案控制台
探索云世界
开发者社区 大数据与机器学习 实时计算 Flink 文章 正文

Flink SQL 问题之执行报错如何解决

2024-02-24 140
版权
版权声明：
本文内容由阿里云实名注册用户自发贡献，版权归原作者所有，阿里云开发者社区不拥有其著作权，亦不承担相应法律责任。具体规则请查看《 阿里云开发者社区用户服务协议》和 《阿里云开发者社区知识产权保护指引》。如果您发现本社区中有涉嫌抄袭的内容，填写 侵权投诉表单进行举报，一经查实，本社区将立刻删除涉嫌侵权内容。
本文涉及的产品
实时计算 Flink 版，5000CU*H 3个月
推荐场景：
实时发现最热Github项目
简介： Flink SQL报错通常指在使用Apache Flink的SQL接口执行数据处理任务时遇到的问题；本合集将收集常见的Flink SQL报错情况及其解决方法，帮助用户迅速恢复数据处理流程。

问题一：flinksql报错Could not find any factory for identifier 'elasticsearch-6' that implements 'org.apache.flink.table.factories.DynamicTableSinkFactory' in the classpath. hello 我在使用flinksql连接器时当我将flink-sql-connector-elasticsearch6_2.11-1.11.1.jar放在lib下，程序正常执行，但是当我在pom中进行配置时会产生如下报错，同样的问题会产生在hbase、jdbc的connector中，请问下这可能是什么造成的


flinksql报错Could not find any factory for identifier 'elasticsearch-6' that implements 'org.apache.flink.table.factories.DynamicTableSinkFactory' in the classpath. hello 我在使用flinksql连接器时当我将flink-sql-connector-elasticsearch6_2.11-1.11.1.jar放在lib下，程序正常执行，但是当我在pom中进行配置时会产生如下报错，同样的问题会产生在hbase、jdbc的connector中，请问下这可能是什么造成的 org.apache.flink.client.program.ProgramInvocationException: The main method caused an error: Unable to create a sink for writing table 'default_catalog.default_database.cloud_behavior_sink'.

Table options are:

'connector'='elasticsearch-6' 'document-type'='cdbp' 'hosts'='http://10.2.11.116:9200;http://10.2.11.117:9200;http://10.2.11.118:9200;http://10.2.11.119:9200' 'index'='flink_sql_test' 'sink.bulk-flush.max-actions'='100' at org.apache.flink.client.program.PackagedProgram.callMainMethod(PackagedProgram.java:302) at org.apache.flink.client.program.PackagedProgram.invokeInteractiveModeForExecution(PackagedProgram.java:198) at org.apache.flink.client.ClientUtils.executeProgram(ClientUtils.java:149) at org.apache.flink.client.cli.CliFrontend.executeProgram(CliFrontend.java:699) at org.apache.flink.client.cli.CliFrontend.run(CliFrontend.java:232) at org.apache.flink.client.cli.CliFrontend.parseParameters(CliFrontend.java:916) at org.apache.flink.client.cli.CliFrontend.lambdamainmainmain10(CliFrontend.java:992) 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:1917) at org.apache.flink.runtime.security.contexts.HadoopSecurityContext.runSecured(HadoopSecurityContext.java:41) at org.apache.flink.client.cli.CliFrontend.main(CliFrontend.java:992) Caused by: org.apache.flink.table.api.ValidationException: Unable to create a sink for writing table 'default_catalog.default_database.cloud_behavior_sink'.

Table options are:

'connector'='elasticsearch-6' 'document-type'='cdbp' 'hosts'='http://10.2.11.116:9200;http://10.2.11.117:9200;http://10.2.11.118:9200;http://10.2.11.119:9200' 'index'='flink_sql_test' 'sink.bulk-flush.max-actions'='100' at org.apache.flink.table.factories.FactoryUtil.createTableSink(FactoryUtil.java:164) at org.apache.flink.table.planner.delegation.PlannerBase.getTableSink(PlannerBase.scala:344) at org.apache.flink.table.planner.delegation.PlannerBase.translateToRel(PlannerBase.scala:204) at org.apache.flink.table.planner.delegation.PlannerBase

anonfun$1.apply(PlannerBase.scala:163)atorg.apache.flink.table.planner.delegation.PlannerBaseanonfun$1.apply(PlannerBase.scala:163)atorg.apache.flink.table.planner.delegation.PlannerBase

anonfun$1.apply(PlannerBase.scala:163) at org.apache.flink.table.planner.delegation.PlannerBaseanonfun1.apply(PlannerBase.scala:163)atscala.collection.TraversableLike1.apply(PlannerBase.scala:163)atscala.collection.TraversableLike1.apply(PlannerBase.scala:163) at scala.collection.TraversableLikeanonfunanonfunanonfunmap1.apply(TraversableLike.scala:234)atscala.collection.TraversableLike1.apply(TraversableLike.scala:234)atscala.collection.TraversableLike1.apply(TraversableLike.scala:234) at scala.collection.TraversableLikeanonfunanonfunanonfunmap1.apply(TraversableLike.scala:234)atscala.collection.Iterator1.apply(TraversableLike.scala:234)atscala.collection.Iterator1.apply(TraversableLike.scala:234) at scala.collection.Iteratorclass.foreach(Iterator.scala:891) at scala.collection.AbstractIterator.foreach(Iterator.scala:1334) at scala.collection.IterableLikeclass.foreach(IterableLike.scala:72)atscala.collection.AbstractIterable.foreach(Iterable.scala:54)atscala.collection.TraversableLikeclass.foreach(IterableLike.scala:72)atscala.collection.AbstractIterable.foreach(Iterable.scala:54)atscala.collection.TraversableLikeclass.foreach(IterableLike.scala:72) at scala.collection.AbstractIterable.foreach(Iterable.scala:54) at scala.collection.TraversableLikeclass.map(TraversableLike.scala:234) at scala.collection.AbstractTraversable.map(Traversable.scala:104) at org.apache.flink.table.planner.delegation.PlannerBase.translate(PlannerBase.scala:163) at org.apache.flink.table.api.internal.TableEnvironmentImpl.translate(TableEnvironmentImpl.java:1264) at org.apache.flink.table.api.internal.TableEnvironmentImpl.executeInternal(TableEnvironmentImpl.java:700) at org.apache.flink.table.api.internal.TableEnvironmentImpl.executeOperation(TableEnvironmentImpl.java:787) at org.apache.flink.table.api.internal.TableEnvironmentImpl.executeSql(TableEnvironmentImpl.java:690) at com.intsig.flink.streaming.streaming_project.common.sql_submit.SqlSubmit.callDML(SqlSubmit.java:97) at com.intsig.flink.streaming.streaming_project.common.sql_submit.SqlSubmit.callCommand(SqlSubmit.java:72) at com.intsig.flink.streaming.streaming_project.common.sql_submit.SqlSubmit.run(SqlSubmit.java:53) at com.intsig.flink.streaming.streaming_project.common.sql_submit.SqlSubmit.main(SqlSubmit.java:24) 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:288) ... 11 more Caused by: org.apache.flink.table.api.ValidationException: Cannot discover a connector using option ''connector'='elasticsearch-6''. at org.apache.flink.table.factories.FactoryUtil.getDynamicTableFactory(FactoryUtil.java:329) at org.apache.flink.table.factories.FactoryUtil.createTableSink(FactoryUtil.java:157) ... 37 more Caused by: org.apache.flink.table.api.ValidationException: Could not find any factory for identifier 'elasticsearch-6' that implements 'org.apache.flink.table.factories.DynamicTableSinkFactory' in the classpath.

Available factory identifiers are:

blackhole hbase-1.4 jdbc kafka print at org.apache.flink.table.factories.FactoryUtil.discoverFactory(FactoryUtil.java:240) at org.apache.flink.table.factories.FactoryUtil.getDynamicTableFactory(FactoryUtil.java:326) ... 38 more*来自志愿者整理的flink


参考回答：

Flink的class加载原则是child first，所以，尽量避免在pom中自己引入flink相关的依赖，避免跟Flink集群环境造成冲突，建议将安装包放在lib下，由flink去加载。


关于本问题的更多回答可点击原文查看：https://developer.aliyun.com/ask/370178?spm=a2c6h.14164896.0.0.42a563bfbbfLHp


问题二：flinksql例子建表的，没有报错就自动结束FAILED了是正常的嘛？


flinksql例子建表的



没有报错就自动结束FAILED了是正常的嘛？


参考回答：

sql也分批和流，如果是批结束且是finish是正常的


关于本问题的更多回答可点击原文查看：https://developer.aliyun.com/ask/500240?spm=a2c6h.14164896.0.0.42a563bfbbfLHp


问题三：使用flinksql mongodb-cdc执行报错怎么办？


使用flinksql mongodb-cdc执行报错unable to create a source for reading table，这个怎么办？


参考回答：

这个错误可能是由于以下原因导致的：

  1. 数据库或集合名称错误：确保你在SQL查询中使用的数据库和集合名称与实际情况相匹配。
  2. 连接字符串错误：确保你在配置CDC连接器时正确指定了MongoDB连接字符串，并且已经在Flink SQL中进行了正确的配置。
  3. 缺少MongoDB驱动程序：如果你没有在Flink的lib目录下放置MongoDB驱动程序，可能会遇到此错误。请确保你已经添加了MongoDB驱动程序，以便FlinkSQL能够与MongoDB数据库建立连接。
  4. CDC版本不兼容：如果你使用的Flink版本和MongoDB CDC版本不兼容，也可能会出现此错误。请尝试使用兼容的版本进行测试。

你可以通过查看具体的错误日志，排除以上可能的原因，并采取相应的措施解决问题。


关于本问题的更多回答可点击原文查看：https://developer.aliyun.com/ask/494471?spm=a2c6h.14164896.0.0.42a563bfbbfLHp


问题四：启动flinksql,然后随便执行一个sql报错，为什么？


启动flinksql,然后随便执行一个sql报错，为什么？



参考回答：

先看一下43268这个端口有没有打开


关于本问题的更多回答可点击原文查看：https://developer.aliyun.com/ask/482512?spm=a2c6h.14164896.0.0.42a563bfbbfLHp


问题五：大佬们Flink 用SQL API sink到 Doris， 中间不做处理，直接CDC 到Doris. job不报错， Doris表里没数据， 这有可能是什么原因呢？


大佬们Flink 用SQL API sink到 Doris， 中间不做处理，直接CDC 到Doris. job不报错， Doris表里没数据， 这有可能是什么原因呢？


参考回答：

打开doris的严格模式


关于本问题的更多回答可点击原文查看：https://developer.aliyun.com/ask/500428?spm=a2c6h.14164896.0.0.42a563bfbbfLHp

文章标签：
实时计算 Flink版
检索分析服务 Elasticsearch版
云数据库 MongoDB 版
流计算
SQL
NoSQL
MongoDB
Java
关键词：
实时计算 Flink版报错
flink SQL
SQL报错
SQL如何解决
SQL flink
相关实践学习
基于Hologres轻松玩转一站式实时仓库
本场景介绍如何利用阿里云MaxCompute、实时计算Flink和交互式分析服务Hologres开发离线、实时数据融合分析的数据大屏应用。
Linux入门到精通
本套课程是从入门开始的Linux学习课程，适合初学者阅读。由浅入深案例丰富，通俗易懂。主要涉及基础的系统操作以及工作中常用的各种服务软件的应用、部署和优化。即使是零基础的学员，只要能够坚持把所有章节都学完，也一定会受益匪浅。
cuicuicuic
目录
相关文章
xiaohua616
|
4天前
|
SQL XML Java
mybatis 调用修改SQL时 出现了一个问题 没有修改成功也没有报错
mybatis 调用修改SQL时 出现了一个问题 没有修改成功也没有报错
xiaohua616
13 0
cuicuicuic
|
1月前
|
SQL JSON Kubernetes
Seata常见问题之服务端 error日志没有输出,客户端执行sql报错如何解决
Seata 是一个开源的分布式事务解决方案，旨在提供高效且简单的事务协调机制，以解决微服务架构下跨服务调用（分布式场景）的一致性问题。以下是Seata常见问题的一个合集
cuicuicuic
107 0
cuicuicuic
|
1月前
|
SQL 存储 Kubernetes
Seata常见问题之mybatisplus的批量插入方法报SQL错误如何解决
Seata 是一个开源的分布式事务解决方案，旨在提供高效且简单的事务协调机制，以解决微服务架构下跨服务调用（分布式场景）的一致性问题。以下是Seata常见问题的一个合集
cuicuicuic
26 0
vohelon
|
1月前
|
Java 关系型数据库 MySQL
Flink1.18.1和CDC2.4.1 本地没问题 提交任务到服务器 报错java.lang.NoClassDefFoundError: Could not initialize class io.debezium.connector.mysql.MySqlConnectorConfig
【2月更文挑战第33天】Flink1.18.1和CDC2.4.1 本地没问题 提交任务到服务器 报错java.lang.NoClassDefFoundError: Could not initialize class io.debezium.connector.mysql.MySqlConnectorConfig
vohelon
52 2
sunrr
|
1月前
|
Java 关系型数据库 MySQL
Flink CDC有见这个报错不？
【2月更文挑战第29天】Flink CDC有见这个报错不？
sunrr
22 2
你鞋带开了~
|
1月前
|
存储 关系型数据库 MySQL
Flink CDC产品常见问题之写hudi的时候报错如何解决
Flink CDC（Change Data Capture）是一个基于Apache Flink的实时数据变更捕获库，用于实现数据库的实时同步和变更流的处理；在本汇总中，我们组织了关于Flink CDC产品在实践中用户经常提出的问题及其解答，目的是辅助用户更好地理解和应用这一技术，优化实时数据处理流程。
你鞋带开了~
406 2
你鞋带开了~
|
1月前
|
资源调度 关系型数据库 测试技术
Flink CDC产品常见问题之没有报错但是一直监听不到数据如何解决
Flink CDC（Change Data Capture）是一个基于Apache Flink的实时数据变更捕获库，用于实现数据库的实时同步和变更流的处理；在本汇总中，我们组织了关于Flink CDC产品在实践中用户经常提出的问题及其解答，目的是辅助用户更好地理解和应用这一技术，优化实时数据处理流程。
你鞋带开了~
43 1
你鞋带开了~
|
1月前
|
缓存 监控 Java
Flink CDC产品常见问题之flink集群jps命令报错如何解决
Flink CDC（Change Data Capture）是一个基于Apache Flink的实时数据变更捕获库，用于实现数据库的实时同步和变更流的处理；在本汇总中，我们组织了关于Flink CDC产品在实践中用户经常提出的问题及其解答，目的是辅助用户更好地理解和应用这一技术，优化实时数据处理流程。
你鞋带开了~
29 1
小小Java开发者
|
1月前
|
SQL 数据库 索引
解决SQL报错：索引中丢失IN或OUT參数
解决SQL报错：索引中丢失IN或OUT參数
小小Java开发者
21 1
三分钟热度的鱼
|
1月前
|
Oracle 关系型数据库 MySQL
Flink CDC产品常见问题之用superset连接starrocks报错如何解决
Flink CDC（Change Data Capture）是一个基于Apache Flink的实时数据变更捕获库，用于实现数据库的实时同步和变更流的处理；在本汇总中，我们组织了关于Flink CDC产品在实践中用户经常提出的问题及其解答，目的是辅助用户更好地理解和应用这一技术，优化实时数据处理流程。
三分钟热度的鱼
21 0
大数据与机器学习

实时计算 Flink

热门文章

最新文章

  • 1
    数据仓库介绍与实时数仓案例
  • 2
    权威详解 | 阿里新一代实时计算引擎 Blink，每秒支持数十亿次计算
  • 3
    OPPO数据中台之基石：基于Flink SQL构建实数据仓库
  • 4
    流计算StreamCompute
  • 5
    阿里云实时计算产品案例&解决方案汇总
  • 6
    回顾 | Kafka x Flink Meetup 与世界人工智能大会大数据 AI 专场精彩回顾（附PPT下载）
  • 7
    Flink SQL 功能解密系列 —— 流式 TopN 挑战与实现
  • 8
    为什么说流处理即未来？
  • 9
    基于实时计算（Flink）打造一个简单的实时推荐系统
  • 10
    Flink Checkpoint 问题排查实用指南
  • 1
    阿里巴巴瓴羊基于 Flink 实时计算的优化和实践
    174
  • 2
    友盟+｜如何通过阿里云Flink+Paimon实现流式湖仓落地方案
    252
  • 3
    官宣｜Apache Paimon 毕业成为顶级项目，数据湖步入实时新篇章！
    719
  • 4
    Flink CDC在阿里云DataWorks数据集成应用实践
    442
  • 5
    Flink+Paimon+Hologres 构建实时湖仓数据分析
    71467
  • 6
    阿里云Flink-自定义kafka format实践及踩坑记录（以protobuf为例）
    770
  • 7
    Flink 流批一体在模型特征场景的使用
    77082
  • 8
    阿里云实时计算Flink的产品化思考与实践【下】
    110835
  • 9
    阿里云实时计算Flink的产品化思考与实践【上】
    3370
  • 10
    官宣｜阿里巴巴捐赠的 Flink CDC 项目正式加入 Apache 基金会
    1416

    • 相关产品

  • 实时计算 Flink版
    文档详情 产品详情

    • 相关课程

    更多
  • 开源 Flink 极速上手教程
  • SQL入门与实践
  • 大数据实时计算框架Spark快速入门
  • Apache Flink 入门
  • 数据库及SQL/MySQL基础
  • SQL进阶及查询

    • 相关电子书

    更多
  • Flink CDC Meetup PPT - 龚中强
  • Flink CDC Meetup PPT - 王赫
  • Flink CDC Meetup PPT - 覃立辉

    • 相关实验场景

    更多
  • SQL的增删改查及函数应用
  • PolarDB MySQL自动SQL限流
  • 如何开通SQL Server的专属集群
  • 使用DAS实现数据库自动SQL优化
  • 使用DAS实现数据库自动SQL限流
  • 一小时快速掌握 SQL 语法
    • 下一篇
    部署LAMP环境（Alibaba Cloud Linux 3）