实时计算 Flink版操作报错合集之同步tidb到hudi报错，一般是什么原因-阿里云开发者社区

实时计算 Flink版操作报错合集之同步tidb到hudi报错，一般是什么原因

2024-07-16 64

版权

本文内容由阿里云实名注册用户自发贡献，版权归原作者所有，阿里云开发者社区不拥有其著作权，亦不承担相应法律责任。具体规则请查看《阿里云开发者社区用户服务协议》和《阿里云开发者社区知识产权保护指引》。如果您发现本社区中有涉嫌抄袭的内容，填写侵权投诉表单进行举报，一经查实，本社区将立刻删除涉嫌侵权内容。

本文涉及的产品

实时计算 Flink 版，5000CU*H 3个月

简介： 在使用实时计算Flink版过程中，可能会遇到各种错误，了解这些错误的原因及解决方法对于高效排错至关重要。针对具体问题，查看Flink的日志是关键，它们通常会提供更详细的错误信息和堆栈跟踪，有助于定位问题。此外，Flink社区文档和官方论坛也是寻求帮助的好去处。以下是一些常见的操作报错及其可能的原因与解决策略。

问题一：Flink中kafka总共有16266条数据, 但是消费到6144就停止并报这两条日志了？

Flink中kafka总共有16266条数据, 但是消费到6144就停止并报这两条日志了？

参考答案：

从你给出的日志来看，问题是关于Kafka消费者未能获取足够的offsets而导致程序挂起。这种现象通常发生在消费者的偏移量设置不正确的情况下。

首先，你需要确认一下这两个Consumer实例所对应的Topic名称以及Group ID是否一致。如果不是的话，可能会出现一个Consumer实例已经读取到了某个特定位置的数据，但另一个却还在等待初始Offset的位置上。

其次，查看下这两个Consumer实例的group.id属性是否相同。因为每个不同的组ID对应的是不同的消息队列分区，所以不同组内的Consumer应该有不同的偏移量设定才能正常工作。

最后，建议你在生产环境中添加一些额外的信息以便于定

关于本问题的更多回答可点击进行查看：

https://developer.aliyun.com/ask/587251

问题二：使用doris 1.2.1版本，没用使用docker安装，结果报这个错误。请问是什么原因啊?

使用doris 1.2.1版本，没用使用docker安装，在跑官网的MySQL 到 Doris的 pipeline时，结果报这个错误。请问是什么原因啊?2024-01-06 11:42:57

java.lang.RuntimeException: Failed to schema change, CreateTableEvent{tableId=app_db.orders, schema=columns={id INT NOT NULL,price DECIMAL(10, 2) NOT NULL}, primaryKeys=id, options=()}, reason: SchemaChange request error with Failed to schemaChange, response: {"msg":"Error","code":1,"data":"Failed to execute sql: org.apache.doris.common.AnalysisException: errCode = 2, detailMessage = Syntax error in line 1:\n...TED BY HASH(id) BUCKETS AUTO PROPERTIES ('replicati...\n ^\nEncountered: IDENTIFIER\nExpected\n","count":0}

at com.ververica.cdc.connectors.doris.sink.DorisMetadataApplier.applySchemaChange(DorisMetadataApplier.java:86)

at com.ververica.cdc.runtime.operators.schema.coordinator.SchemaRegistryRequestHandler.applySchemaChange(SchemaRegistryRequestHandler.java:82)

at com.ververica.cdc.runtime.operators.schema.coordinator.SchemaRegistryRequestHandler.flushSuccess(SchemaRegistryRequestHandler.java:149)

at com.ververica.cdc.runtime.operators.schema.coordinator.SchemaRegistry.handleEventFromOperator(SchemaRegistry.java:123)

at org.apache.flink.runtime.operators.coordination.OperatorCoordinatorHolder.handleEventFromOperator(OperatorCoordinatorHolder.java:204)

at org.apache.flink.runtime.scheduler.DefaultOperatorCoordinatorHandler.deliverOperatorEventToCoordinator(DefaultOperatorCoordinatorHandler.java:121)

at org.apache.flink.runtime.scheduler.SchedulerBase.deliverOperatorEventToCoordinator(SchedulerBase.java:1062)

at org.apache.flink.runtime.jobmaster.JobMaster.sendOperatorEventToCoordinator(JobMaster.java:604)

at sun.reflect.GeneratedMethodAccessor85.invoke(Unknown Source)

at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)

at java.lang.reflect.Method.invoke(Method.java:498)

at org.apache.flink.runtime.rpc.pekko.PekkoRpcActor.lambda$handleRpcInvocation$1(PekkoRpcActor.java:309)

at org.apache.flink.runtime.concurrent.ClassLoadingUtils.runWithContextClassLoader(ClassLoadingUtils.java:83)

at org.apache.flink.runtime.rpc.pekko.PekkoRpcActor.handleRpcInvocation(PekkoRpcActor.java:307)

at org.apache.flink.runtime.rpc.pekko.PekkoRpcActor.handleRpcMessage(PekkoRpcActor.java:222)

at org.apache.flink.runtime.rpc.pekko.FencedPekkoRpcActor.handleRpcMessage(FencedPekkoRpcActor.java:85)

at org.apache.flink.runtime.rpc.pekko.PekkoRpcActor.handleMessage(PekkoRpcActor.java:168)

at org.apache.pekko.japi.pf.UnitCaseStatement.apply(CaseStatements.scala:33)

at org.apache.pekko.japi.pf.UnitCaseStatement.apply(CaseStatements.scala:29)

at scala.PartialFunction.applyOrElse(PartialFunction.scala:127)

at scala.PartialFunction.applyOrElse$(PartialFunction.scala:126)

at org.apache.pekko.japi.pf.UnitCaseStatement.applyOrElse(CaseStatements.scala:29)

at scala.PartialFunction$OrElse.applyOrElse(PartialFunction.scala:175)

at scala.PartialFunction$OrElse.applyOrElse(PartialFunction.scala:176)

at org.apache.pekko.actor.Actor.aroundReceive(Actor.scala:547)

at org.apache.pekko.actor.Actor.aroundReceive$(Actor.scala:545)

at org.apache.pekko.actor.AbstractActor.aroundReceive(AbstractActor.scala:229)

at org.apache.pekko.actor.ActorCell.receiveMessage(ActorCell.scala:590)

at org.apache.pekko.actor.ActorCell.invoke(ActorCell.scala:557)

at org.apache.pekko.dispatch.Mailbox.processMailbox(Mailbox.scala:280)

at org.apache.pekko.dispatch.Mailbox.run(Mailbox.scala:241)

at org.apache.pekko.dispatch.Mailbox.exec(Mailbox.scala:253)

at java.util.concurrent.ForkJoinTask.doExec(ForkJoinTask.java:289)

at java.util.concurrent.ForkJoinPool$WorkQueue.runTask(ForkJoinPool.java:1056)

at java.util.concurrent.ForkJoinPool.runWorker(ForkJoinPool.java:1692)

at java.util.concurrent.ForkJoinWorkerThread.run(ForkJoinWorkerThread.java:157)

参考答案：

语法错误 detailMessage = Syntax error in line 1:\n...TED BY HASH(id) BUCKETS AUTO PROPERTIES ('replicati...\n ^\nEncountered: IDENTIFIER\nExpected\n","count":0}

关于本问题的更多回答可点击进行查看：

https://developer.aliyun.com/ask/587227

问题三：Flink CDC有没有遇到过类似的问题？

Flink CDC有没有大佬遇到过类似的问题，第一次提交任务正常，第二次提交就No suitable driver found for jdbc:mysql了，Flink版本1.16.1，求大佬们指点?

参考答案：

这个问题可能是由于在第二次提交任务时，Flink集群没有正确加载MySQL的JDBC驱动。具体来说，可能的原因有：

在第一次提交任务后，MySQL JDBC驱动没有被正确卸载，导致第二次提交任务时，Flink集群找不到合适的驱动。
Flink集群在启动时没有加载MySQL的JDBC驱动。这可能是因为连接池没有指定驱动，或者在启动时没有正确加载驱动。

为了解决这个问题，你可以尝试以下方法：

确保在Flink集群的所有节点上都已经安装了MySQL的JDBC驱动，并且版本一致。你可以检查Flink的lib目录下是否包含正确的JDBC驱动文件，例如mysql-connector-java-6.0.2-bin.jar。
如果问题仍然存在，你可以尝试在Flink作业的代码中显式地加载MySQL的JDBC驱动。你可以在作业的静态方法中手动加载驱动，以确保在作业启动时能够找到驱动。

关于本问题的更多回答可点击进行查看：

https://developer.aliyun.com/ask/587221

问题四：flinkcdc 同步tidb到hudi报错，哪位遇到过这个问题吗?

flinkcdc 同步tidb到hudi报错，哪位大佬遇到过这个问题吗?org.tikv.shade.io.grpc.StatusRuntimeException: UNAVAILABLE: Keepalive failed. The connection is likely gone

at org.tikv.shade.io.grpc.stub.ClientCalls.toStatusRuntimeException(ClientCalls.java:287) ~[tikv-client-java-3.2.0.jar:na]

at org.tikv.shade.io.grpc.stub.ClientCalls.getUnchecked(ClientCalls.java:268) ~[tikv-client-java-3.2.0.jar:na]

at org.tikv.shade.io.grpc.stub.ClientCalls.blockingUnaryCall(ClientCalls.java:175) ~[tikv-client-java-3.2.0.jar:na]

at org.tikv.kvproto.PDGrpc$PDBlockingStub.getMembers(PDGrpc.java:1868) ~[tikv-client-java-3.2.0.jar:na]

at org.tikv.common.PDClient.getMembers(PDClient.java:443) [tikv-client-java-3.2.0.jar:na]

at org.tikv.common.PDClient.tryUpdateLeader(PDClient.java:565) [tikv-client-java-3.2.0.jar:na]

at org.tikv.common.PDClient.lambda$initCluster$15(PDClient.java:730) [tikv-client-java-3.2.0.jar:na]

at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511) ~[na:1.8.0_381]

at java.util.concurrent.FutureTask.runAndReset$$$capture(FutureTask.java:308) ~[na:1.8.0_381]

at java.util.concurrent.FutureTask.runAndReset(FutureTask.java) ~[na:1.8.0_381]

at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.access$301(ScheduledThreadPoolExecutor.java:180) ~[na:1.8.0_381]

at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:294) ~[na:1.8.0_381]

参考答案：

这个问题可能是由于网络不稳定或者TiDB集群的负载过高导致的。你可以尝试以下方法解决这个问题：

检查网络连接是否正常，确保Flink CDC和TiDB之间的网络通信畅通。
增加TiDB集群的节点数量，以提高集群的处理能力。
调整Flink CDC的配置参数，例如增加任务重试次数、减少任务执行间隔等，以应对高负载情况。
使用监控工具（如Prometheus、Grafana等）对TiDB集群进行实时监控，以便及时发现并解决问题。

关于本问题的更多回答可点击进行查看：

https://developer.aliyun.com/ask/587220

问题五：Flink CDC这是什么问题啊？

Flink CDC这是什么问题啊？

参考答案：

这个问题可能是由于在第二次提交任务时，Flink集群没有正确加载MySQL的JDBC驱动。你可以尝试以下方法解决这个问题：

确保在Flink集群的所有节点上都已经安装了MySQL的JDBC驱动，并且版本一致。你可以检查Flink的lib目录下是否包含正确的JDBC驱动文件，例如mysql-connector-java-6.0.2-bin.jar。
如果问题仍然存在，你可以尝试在Flink作业的代码中显式地加载MySQL的JDBC驱动。你可以在作业的静态方法中手动加载驱动，以确保在作业启动时能够找到驱动。示例代码如下：

public static void main(String[] args) throws Exception {
    // 加载MySQL JDBC驱动
    Class.forName("com.mysql.cj.jdbc.Driver");
    // 你的其他Flink作业代码
}

如果以上方法都无法解决问题，建议查看Flink日志以获取更多详细信息，以便进一步排查问题。

关于本问题的更多回答可点击进行查看：

https://developer.aliyun.com/ask/587219

实时计算 Flink版操作报错合集之同步tidb到hudi报错，一般是什么原因

问题一：Flink中kafka总共有16266条数据, 但是消费到6144就停止并报这两条日志了？