Paimon Flink本地读取数据报错
1.idea本地读取paimon
用idea在本地读取paimon的表时需要添加的依赖除了官网提出的和hadoop相关的，paimon-flink之类相关的除外还需要其他额外依赖
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;public class Test {public static void main(String[] args) {StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);tableEnv.executeSql("CREATE CATALOG my_jdbc WITH (\n" +"'type' = 'paimon',\n" +"'metastore' = 'jdbc',\n" +"'uri' = 'jdbc:mysql://test3:3306/paimon',\n" +"'jdbc.user' = 'root',\n" +"'jdbc.password' = 'N^57F9m2RI#rp8',\n" +"'catalog-key'='jdbc',\n" +"'warehouse' = 'hdfs://test1:8020/user/paimon'\n" +" )");tableEnv.executeSql("use catalog my_jdbc");tableEnv.sqlQuery("select * from test").execute().print();}
}
以上是我的代码，用的jdbc catalog读取hadoop上的数据，下面是最开始的依赖：
<dependency><groupId>org.apache.paimon</groupId><artifactId>paimon-flink-1.18</artifactId><version>0.8.0</version></dependency><dependency><groupId>org.apache.hadoop</groupId><artifactId>hadoop-client</artifactId><version>3.3.6</version></dependency><dependency><groupId>mysql</groupId><artifactId>mysql-connector-java</artifactId><version>8.0.28</version></dependency><dependency><groupId>org.apache.flink</groupId><artifactId>flink-core</artifactId><version>1.18.0</version></dependency><!-- https://mvnrepository.com/artifact/org.apache.flink/flink-table-planner --><dependency><groupId>org.apache.flink</groupId><artifactId>flink-table-planner_2.12</artifactId><version>1.18.0</version></dependency><!-- https://mvnrepository.com/artifact/org.apache.flink/flink-table-api-java --><dependency><groupId>org.apache.flink</groupId><artifactId>flink-table-api-java</artifactId><version>1.18.0</version></dependency><dependency><groupId>org.apache.flink</groupId><artifactId>flink-clients</artifactId><version>1.18.0</version></dependency>2. 报错1
Exception in thread "main" java.lang.NoClassDefFoundError: org/apache/flink/connector/base/source/hybrid/HybridSource$SourceFactory
Exception in thread "main" java.lang.NoClassDefFoundError: org/apache/flink/connector/base/source/hybrid/HybridSource$SourceFactoryat org.apache.paimon.flink.source.DataTableSource.getScanRuntimeProvider(DataTableSource.java:192)at org.apache.paimon.flink.source.table.BaseTableSource.getScanRuntimeProvider(BaseTableSource.java:42)at org.apache.flink.table.planner.connectors.DynamicSourceUtils.validateScanSource(DynamicSourceUtils.java:478)at org.apache.flink.table.planner.connectors.DynamicSourceUtils.prepareDynamicSource(DynamicSourceUtils.java:161)at org.apache.flink.table.planner.connectors.DynamicSourceUtils.convertSourceToRel(DynamicSourceUtils.java:125)at org.apache.flink.table.planner.plan.schema.CatalogSourceTable.toRel(CatalogSourceTable.java:118)at org.apache.calcite.sql2rel.SqlToRelConverter.toRel(SqlToRelConverter.java:3997)at org.apache.calcite.sql2rel.SqlToRelConverter.convertIdentifier(SqlToRelConverter.java:2867)at org.apache.calcite.sql2rel.SqlToRelConverter.convertFrom(SqlToRelConverter.java:2427)at org.apache.calcite.sql2rel.SqlToRelConverter.convertFrom(SqlToRelConverter.java:2341)at org.apache.calcite.sql2rel.SqlToRelConverter.convertFrom(SqlToRelConverter.java:2286)at org.apache.calcite.sql2rel.SqlToRelConverter.convertSelectImpl(SqlToRelConverter.java:723)at org.apache.calcite.sql2rel.SqlToRelConverter.convertSelect(SqlToRelConverter.java:709)at org.apache.calcite.sql2rel.SqlToRelConverter.convertQueryRecursive(SqlToRelConverter.java:3843)at org.apache.calcite.sql2rel.SqlToRelConverter.convertQuery(SqlToRelConverter.java:617)at org.apache.flink.table.planner.calcite.FlinkPlannerImpl.org$apache$flink$table$planner$calcite$FlinkPlannerImpl$$rel(FlinkPlannerImpl.scala:229)at org.apache.flink.table.planner.calcite.FlinkPlannerImpl.rel(FlinkPlannerImpl.scala:205)at org.apache.flink.table.planner.operations.SqlNodeConvertContext.toRelRoot(SqlNodeConvertContext.java:69)at org.apache.flink.table.planner.operations.converters.SqlQueryConverter.convertSqlNode(SqlQueryConverter.java:48)at org.apache.flink.table.planner.operations.converters.SqlNodeConverters.convertSqlNode(SqlNodeConverters.java:73)at org.apache.flink.table.planner.operations.SqlNodeToOperationConversion.convertValidatedSqlNode(SqlNodeToOperationConversion.java:272)at org.apache.flink.table.planner.operations.SqlNodeToOperationConversion.convert(SqlNodeToOperationConversion.java:262)at org.apache.flink.table.planner.delegation.ParserImpl.parse(ParserImpl.java:106)at org.apache.flink.table.api.internal.TableEnvironmentImpl.sqlQuery(TableEnvironmentImpl.java:708)at kafka.Test.main(Test.java:25)
Caused by: java.lang.ClassNotFoundException: org.apache.flink.connector.base.source.hybrid.HybridSource$SourceFactoryat java.net.URLClassLoader.findClass(URLClassLoader.java:387)at java.lang.ClassLoader.loadClass(ClassLoader.java:418)at sun.misc.Launcher$AppClassLoader.loadClass(Launcher.java:352)at java.lang.ClassLoader.loadClass(ClassLoader.java:351)... 25 more从报错大概可以看出需要这个 org/apache/flink/connector/base路径底下的source/hybrid/HybridSource$SourceFactory类，查了一下这个类在flink-connector-base类里面，所以添加上这个依赖：
<dependency><groupId>org.apache.flink</groupId><artifactId>flink-connector-base</artifactId><version>1.18.0</version> </dependency>
3.报错2
加上了上面的依赖继续跑然后又报错了：NoClassDefFoundError: org/apache/flink/connector/file/src/reader/BulkFormat$RecordIterator
Exception in thread "main" java.lang.RuntimeException: Failed to fetch next resultat org.apache.flink.streaming.api.operators.collect.CollectResultIterator.nextResultFromFetcher(CollectResultIterator.java:109)at org.apache.flink.streaming.api.operators.collect.CollectResultIterator.hasNext(CollectResultIterator.java:80)at org.apache.flink.table.planner.connectors.CollectDynamicSink$CloseableRowIteratorWrapper.hasNext(CollectDynamicSink.java:222)at org.apache.flink.table.utils.print.TableauStyle.print(TableauStyle.java:120)at org.apache.flink.table.api.internal.TableResultImpl.print(TableResultImpl.java:153)at kafka.Test.main(Test.java:25)
Caused by: java.io.IOException: Failed to fetch job execution resultat org.apache.flink.streaming.api.operators.collect.CollectResultFetcher.getAccumulatorResults(CollectResultFetcher.java:185)at org.apache.flink.streaming.api.operators.collect.CollectResultFetcher.next(CollectResultFetcher.java:121)at org.apache.flink.streaming.api.operators.collect.CollectResultIterator.nextResultFromFetcher(CollectResultIterator.java:106)... 5 more
Caused by: java.util.concurrent.ExecutionException: org.apache.flink.runtime.client.JobExecutionException: Job execution failed.at java.util.concurrent.CompletableFuture.reportGet(CompletableFuture.java:357)at java.util.concurrent.CompletableFuture.get(CompletableFuture.java:1928)at org.apache.flink.streaming.api.operators.collect.CollectResultFetcher.getAccumulatorResults(CollectResultFetcher.java:183)... 7 more
Caused by: org.apache.flink.runtime.client.JobExecutionException: Job execution failed.at org.apache.flink.runtime.jobmaster.JobResult.toJobExecutionResult(JobResult.java:144)at org.apache.flink.runtime.minicluster.MiniClusterJobClient.lambda$getJobExecutionResult$3(MiniClusterJobClient.java:141)at java.util.concurrent.CompletableFuture.uniApply(CompletableFuture.java:616)at java.util.concurrent.CompletableFuture.uniApplyStage(CompletableFuture.java:628)at java.util.concurrent.CompletableFuture.thenApply(CompletableFuture.java:1996)at org.apache.flink.runtime.minicluster.MiniClusterJobClient.getJobExecutionResult(MiniClusterJobClient.java:138)at org.apache.flink.streaming.api.operators.collect.CollectResultFetcher.getAccumulatorResults(CollectResultFetcher.java:182)... 7 more
Caused by: org.apache.flink.runtime.JobException: Recovery is suppressed by NoRestartBackoffTimeStrategyat org.apache.flink.runtime.executiongraph.failover.flip1.ExecutionFailureHandler.handleFailure(ExecutionFailureHandler.java:176)at org.apache.flink.runtime.executiongraph.failover.flip1.ExecutionFailureHandler.getFailureHandlingResult(ExecutionFailureHandler.java:107)at org.apache.flink.runtime.scheduler.DefaultScheduler.recordTaskFailure(DefaultScheduler.java:285)at org.apache.flink.runtime.scheduler.DefaultScheduler.handleTaskFailure(DefaultScheduler.java:276)at org.apache.flink.runtime.scheduler.DefaultScheduler.onTaskFailed(DefaultScheduler.java:269)at org.apache.flink.runtime.scheduler.SchedulerBase.onTaskExecutionStateUpdate(SchedulerBase.java:764)at org.apache.flink.runtime.scheduler.SchedulerBase.updateTaskExecutionState(SchedulerBase.java:741)at org.apache.flink.runtime.scheduler.SchedulerNG.updateTaskExecutionState(SchedulerNG.java:83)at org.apache.flink.runtime.jobmaster.JobMaster.updateTaskExecutionState(JobMaster.java:488)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.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 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:175)
Caused by: java.lang.BootstrapMethodError: java.lang.NoClassDefFoundError: org/apache/flink/connector/file/src/reader/BulkFormat$RecordIteratorat org.apache.paimon.flink.source.FileStoreSourceReader.<init>(FileStoreSourceReader.java:55)at org.apache.paimon.flink.source.FlinkSource.createReader(FlinkSource.java:59)at org.apache.flink.streaming.api.operators.SourceOperator.initReader(SourceOperator.java:314)at org.apache.flink.streaming.runtime.tasks.SourceOperatorStreamTask.init(SourceOperatorStreamTask.java:93)at org.apache.flink.streaming.runtime.tasks.StreamTask.restoreInternal(StreamTask.java:718)at org.apache.flink.streaming.runtime.tasks.StreamTask.restore(StreamTask.java:693)at org.apache.flink.runtime.taskmanager.Task.runWithSystemExitMonitoring(Task.java:953)at org.apache.flink.runtime.taskmanager.Task.restoreAndInvoke(Task.java:922)at org.apache.flink.runtime.taskmanager.Task.doRun(Task.java:746)at org.apache.flink.runtime.taskmanager.Task.run(Task.java:562)at java.lang.Thread.run(Thread.java:750)
Caused by: java.lang.NoClassDefFoundError: org/apache/flink/connector/file/src/reader/BulkFormat$RecordIterator... 11 more
Caused by: java.lang.ClassNotFoundException: org.apache.flink.connector.file.src.reader.BulkFormat$RecordIteratorat java.net.URLClassLoader.findClass(URLClassLoader.java:387)at java.lang.ClassLoader.loadClass(ClassLoader.java:418)at sun.misc.Launcher$AppClassLoader.loadClass(Launcher.java:352)at java.lang.ClassLoader.loadClass(ClassLoader.java:351)... 11 more从上面大概可以看出应该是缺flink-connector-files依赖，添加上去之后成功了
<dependency><groupId>org.apache.flink</groupId><artifactId>flink-connector-files</artifactId><version>1.18-SNAPSHOT</version> </dependency>
