flink tableapi:未能获取

neskvpey  于 2021-07-15  发布在  Flink
关注(0)|答案(0)|浏览(1063)

我现在正在玩flink tableapi。我创建了一个表，想看看表中有什么。我想如果我使用select语句和.collect（），我可以得到一个迭代器，我可以用它来迭代并查看表中的内容。但它给了我错误。
代码：

  1. public static void main(final String[] args) throws Exception {
  3.         // initialize table environment
  4.         StreamExecutionEnvironment bsEnv = StreamExecutionEnvironment.getExecutionEnvironment();
  5.         EnvironmentSettings bsSettings = EnvironmentSettings.newInstance().useBlinkPlanner().inStreamingMode().build();
  6.         StreamTableEnvironment tableEnv = StreamTableEnvironment.create(bsEnv, bsSettings);
  8.         // A table representing some machines
  9.         tableEnv.executeSql("CREATE TEMPORARY TABLE machines(id INT) WITH ( 'connector' = 'kafka', " +
  10.                 "'topic' = 'quickstart-events', 'properties.bootstrap.servers' = 'localhost:9092', 'value.format' = 'json')");
  12.         // Add four machines
  13.         tableEnv.executeSql("INSERT INTO machines values(1)");
  14.         tableEnv.executeSql("INSERT INTO machines values(3)");
  15.         tableEnv.executeSql("INSERT INTO machines values(5)");
  16.         tableEnv.executeSql("INSERT INTO machines values(8)");
  18.         final CloseableIterator<Row> fetch = tableEnv.executeSql("SELECT id FROM machines").collect();
  20.         while (fetch.hasNext()) {
  21.             System.out.println(fetch.next());
  22.         }
  23. }

我遇到的错误：

  1. Exception in thread "main" java.lang.RuntimeException: Failed to fetch next result
  2.     at org.apache.flink.streaming.api.operators.collect.CollectResultIterator.nextResultFromFetcher(CollectResultIterator.java:106)
  3.     at org.apache.flink.streaming.api.operators.collect.CollectResultIterator.hasNext(CollectResultIterator.java:77)
  4.     at org.apache.flink.table.planner.sinks.SelectTableSinkBase$RowIteratorWrapper.hasNext(SelectTableSinkBase.java:115)
  5.     at org.apache.flink.table.api.internal.TableResultImpl$CloseableRowIteratorWrapper.hasNext(TableResultImpl.java:355)
  6.     at com.vmware.dcm.examples.QuickStartWithTableAPI.main(QuickStartWithTableAPI.java:58)
  7. Caused by: java.io.IOException: Failed to fetch job execution result
  8.     at org.apache.flink.streaming.api.operators.collect.CollectResultFetcher.getAccumulatorResults(CollectResultFetcher.java:175)
  9.     at org.apache.flink.streaming.api.operators.collect.CollectResultFetcher.next(CollectResultFetcher.java:126)
  10.     at org.apache.flink.streaming.api.operators.collect.CollectResultIterator.nextResultFromFetcher(CollectResultIterator.java:103)
  11. Caused by: java.io.IOException: Failed to fetch job execution result
  13.     ... 4 more
  14. Caused by: java.util.concurrent.ExecutionException: org.apache.flink.runtime.client.JobExecutionException: Job execution failed.
  15.     at java.base/java.util.concurrent.CompletableFuture.reportGet(CompletableFuture.java:395)
  16.     at java.base/java.util.concurrent.CompletableFuture.get(CompletableFuture.java:2093)
  17. Caused by: java.util.concurrent.ExecutionException: org.apache.flink.runtime.client.JobExecutionException: Job execution failed.
  19.     at org.apache.flink.streaming.api.operators.collect.CollectResultFetcher.getAccumulatorResults(CollectResultFetcher.java:172)
  20.     ... 6 more
  21. Caused by: org.apache.flink.runtime.client.JobExecutionException: Job execution failed.
  22.     at org.apache.flink.runtime.jobmaster.JobResult.toJobExecutionResult(JobResult.java:147)
  23.     at org.apache.flink.runtime.minicluster.MiniClusterJobClient.lambda$getJobExecutionResult$2(MiniClusterJobClient.java:119)
  24. Caused by: org.apache.flink.runtime.client.JobExecutionException: Job execution failed.
  26.     at java.base/java.util.concurrent.CompletableFuture.uniApplyNow(CompletableFuture.java:680)
  27.     at java.base/java.util.concurrent.CompletableFuture.uniApplyStage(CompletableFuture.java:658)
  28.     at java.base/java.util.concurrent.CompletableFuture.thenApply(CompletableFuture.java:2165)
  29.     at org.apache.flink.runtime.minicluster.MiniClusterJobClient.getJobExecutionResult(MiniClusterJobClient.java:117)
  30.     ... 7 more
  31. Caused by: org.apache.flink.runtime.JobException: Recovery is suppressed by NoRestartBackoffTimeStrategy
  32.     at org.apache.flink.runtime.executiongraph.failover.flip1.ExecutionFailureHandler.handleFailure(ExecutionFailureHandler.java:116)
  33. Caused by: org.apache.flink.runtime.JobException: Recovery is suppressed by NoRestartBackoffTimeStrategy
  35.     at org.apache.flink.runtime.executiongraph.failover.flip1.ExecutionFailureHandler.getFailureHandlingResult(ExecutionFailureHandler.java:78)
  36.     at org.apache.flink.runtime.scheduler.DefaultScheduler.handleTaskFailure(DefaultScheduler.java:224)
  37.     at org.apache.flink.runtime.scheduler.DefaultScheduler.maybeHandleTaskFailure(DefaultScheduler.java:217)
  38.     at org.apache.flink.runtime.scheduler.DefaultScheduler.updateTaskExecutionStateInternal(DefaultScheduler.java:208)
  39.     at org.apache.flink.runtime.scheduler.SchedulerBase.updateTaskExecutionState(SchedulerBase.java:610)
  40.     at org.apache.flink.runtime.scheduler.SchedulerNG.updateTaskExecutionState(SchedulerNG.java:89)
  41.     at org.apache.flink.runtime.jobmaster.JobMaster.updateTaskExecutionState(JobMaster.java:419)
  42.     at jdk.internal.reflect.GeneratedMethodAccessor49.invoke(Unknown Source)
  43.     at java.base/jdk.internal.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
  44.     at java.base/java.lang.reflect.Method.invoke(Method.java:567)
  45.     at org.apache.flink.runtime.rpc.akka.AkkaRpcActor.handleRpcInvocation(AkkaRpcActor.java:286)
  46.     at org.apache.flink.runtime.rpc.akka.AkkaRpcActor.handleRpcMessage(AkkaRpcActor.java:201)
  47.     at org.apache.flink.runtime.rpc.akka.FencedAkkaRpcActor.handleRpcMessage(FencedAkkaRpcActor.java:74)
  48.     at org.apache.flink.runtime.rpc.akka.AkkaRpcActor.handleMessage(AkkaRpcActor.java:154)
  49.     at akka.japi.pf.UnitCaseStatement.apply(CaseStatements.scala:26)
  50.     at akka.japi.pf.UnitCaseStatement.apply(CaseStatements.scala:21)
  51.     at scala.PartialFunction$class.applyOrElse(PartialFunction.scala:123)
  52.     at akka.japi.pf.UnitCaseStatement.applyOrElse(CaseStatements.scala:21)
  53.     at scala.PartialFunction$OrElse.applyOrElse(PartialFunction.scala:170)
  54.     at scala.PartialFunction$OrElse.applyOrElse(PartialFunction.scala:171)
  55.     at scala.PartialFunction$OrElse.applyOrElse(PartialFunction.scala:171)
  56.     at akka.actor.Actor$class.aroundReceive(Actor.scala:517)
  57.     at akka.actor.AbstractActor.aroundReceive(AbstractActor.scala:225)
  58.     at akka.actor.ActorCell.receiveMessage(ActorCell.scala:592)
  59.     at akka.actor.ActorCell.invoke(ActorCell.scala:561)
  60.     at akka.dispatch.Mailbox.processMailbox(Mailbox.scala:258)
  61.     at akka.dispatch.Mailbox.run(Mailbox.scala:225)
  62.     at akka.dispatch.Mailbox.exec(Mailbox.scala:235)
  63.     at akka.dispatch.forkjoin.ForkJoinTask.doExec(ForkJoinTask.java:260)
  64.     at akka.dispatch.forkjoin.ForkJoinPool$WorkQueue.runTask(ForkJoinPool.java:1339)
  65.     at akka.dispatch.forkjoin.ForkJoinPool.runWorker(ForkJoinPool.java:1979)
  66.     at akka.dispatch.forkjoin.ForkJoinWorkerThread.run(ForkJoinWorkerThread.java:107)
  67. Caused by: org.apache.kafka.common.errors.TimeoutException: Timeout expired while fetching topic metadata
  68. Caused by: org.apache.kafka.common.errors.TimeoutException: Timeout expired while fetching topic metadata

我该怎么解决这个问题？

apache-flinkflink-sqlflink-streaming

暂无答案！

目前还没有任何答案，快来回答吧！

我来回答

相关问题

    查看更多

    热门标签

    更多
    JavaquerypythonNode开发语言requestUtil数据库Table后端算法LoggerMessageElementParser

    最新问答

    更多
    • 蜀ICP备13028337号-1 大数据知识库 https://www.saoniuhuo.com © All rights reserved
    • 本站内容来源互联网,如果侵犯您的权益请联系我们删除, 联系方式：448109455@qq.com