登录
当前位置： 首页 > 工具软件 > Benv > 使用案例 >

flink shell的local模式(benv与senv的使用+处理报错的解决方案)

陆翔飞
2023-12-01

启动

start-scala-shell.sh local

下面是benv和senv的例子[1]

flnk shell变量用途环境变量的作用
benvDataSet输出到文件中
senvDataStream打印到屏幕上

#############################################################################################

DataSet[1]

val text = benv.fromElements("To be, or not to be,--that is the question:--","Whether 'tis nobler in the mind to suffer","The slings and arrows of outrageous fortune","Or to take arms against a sea of troubles,")
val counts = text
val counts = text.flatMap { _.toLowerCase.split("\\W+") }.map { (_, 1) }.groupBy(0).sum(1)
counts.print()
counts.writeAsText("/tmp/out1")
benv.execute("batch test")

(Python3.6) appleyuchi@Desktop:~$ cd /tmp/out1
(Python3.6) appleyuchi@Desktop:out1$ ls
1  2  3  4
(Python3.6) appleyuchi@Desktop:out1$ cat 1
(and,1)
(arrows,1)
(be,2)
(is,1)
(nobler,1)
(of,2)
(Python3.6) appleyuchi@Desktop:out1$ cat 2
(a,1)
(in,1)
(mind,1)
(or,2)
(slings,1)
(suffer,1)
(Python3.6) appleyuchi@Desktop:out1$ cat 3
(against,1)
(arms,1)
(not,1)
(outrageous,1)
(sea,1)
(the,3)
(tis,1)
(troubles,1)
(whether,1)
(Python3.6) appleyuchi@Desktop:out1$ cat 4
(fortune,1)
(question,1)
(take,1)
(that,1)
(to,4)
 

#####################################################################################################

DataStream[1]

val textStreaming = senv.fromElements("To be, or not to be,--that is the question:--")
val countsStreaming = textStreaming.flatMap { _.toLowerCase.split("\\W+") }.map { (_, 1) }.keyBy(0).sum(1)
countsStreaming.print()
senv.execute("Streaming Wordcount")
(to,1)
(be,1)
(or,1)
(not,1)
(to,2)
(be,2)
(that,1)
(is,1)
(the,1)
(question,1)

注意，上面这个如果没有countsStreaming.print(),senv.execute("Streaming Wordcount")就无法输出结果

 

#####################################-附录-报错处理1######################################

如果出现报错:

Exception in thread "main" org.apache.flink.util.FlinkException: Could not create the DispatcherResourceManagerComponent.
单机模式显然没啥资源好分配的，肯定是没关干净。

(Python3.6) appleyuchi@Desktop:bin$ jps
10065 NodeManager
9346 DataNode
8099 Jps
9882 ResourceManager
9178 NameNode
8010 FlinkShell
5053 FlinkShell

kill -9 8010 5053

然后重启flink shell即可

#####################################-附录-报错处理2######################################

如果出现下面的报错

scala> senv.execute("My streaming program")
org.apache.flink.client.program.ProgramInvocationException: Job failed (JobID: ec2d3e37b4e2be70508f8f9da63ae13b)
  at org.apache.flink.client.deployment.ClusterClientJobClientAdapter.lambda$null$6(ClusterClientJobClientAdapter.java:112)
  at java.util.concurrent.CompletableFuture.uniApply(CompletableFuture.java:602)
  at java.util.concurrent.CompletableFuture$UniApply.tryFire(CompletableFuture.java:577)
  at java.util.concurrent.CompletableFuture.postComplete(CompletableFuture.java:474)
  at java.util.concurrent.CompletableFuture.complete(CompletableFuture.java:1962)
  at org.apache.flink.client.program.rest.RestClusterClient.lambda$pollResourceAsync$21(RestClusterClient.java:565)
  at java.util.concurrent.CompletableFuture.uniWhenComplete(CompletableFuture.java:760)
  at java.util.concurrent.CompletableFuture$UniWhenComplete.tryFire(CompletableFuture.java:736)
  at java.util.concurrent.CompletableFuture.postComplete(CompletableFuture.java:474)
  at java.util.concurrent.CompletableFuture.complete(CompletableFuture.java:1962)
  at org.apache.flink.runtime.concurrent.FutureUtils.lambda$retryOperationWithDelay$8(FutureUtils.java:291)
  at java.util.concurrent.CompletableFuture.uniWhenComplete(CompletableFuture.java:760)
  at java.util.concurrent.CompletableFuture$UniWhenComplete.tryFire(CompletableFuture.java:736)
  at java.util.concurrent.CompletableFuture.postComplete(CompletableFuture.java:474)
  at java.util.concurrent.CompletableFuture.postFire(CompletableFuture.java:561)
  at java.util.concurrent.CompletableFuture$UniCompose.tryFire(CompletableFuture.java:929)
  at java.util.concurrent.CompletableFuture$Completion.run(CompletableFuture.java:442)
  at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142)
  at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)
  at java.lang.Thread.run(Thread.java:748)
Caused by: org.apache.flink.runtime.client.JobExecutionException: Job execution failed.
  at org.apache.flink.runtime.jobmaster.JobResult.toJobExecutionResult(JobResult.java:147)
  at org.apache.flink.client.deployment.ClusterClientJobClientAdapter.lambda$null$6(ClusterClientJobClientAdapter.java:110)
  ... 19 more
Caused by: org.apache.flink.runtime.JobException: Recovery is suppressed by NoRestartBackoffTimeStrategy
  at org.apache.flink.runtime.executiongraph.failover.flip1.ExecutionFailureHandler.handleFailure(ExecutionFailureHandler.java:110)
  at org.apache.flink.runtime.executiongraph.failover.flip1.ExecutionFailureHandler.getFailureHandlingResult(ExecutionFailureHandler.java:76)
  at org.apache.flink.runtime.scheduler.DefaultScheduler.handleTaskFailure(DefaultScheduler.java:192)
  at org.apache.flink.runtime.scheduler.DefaultScheduler.maybeHandleTaskFailure(DefaultScheduler.java:186)
  at org.apache.flink.runtime.scheduler.DefaultScheduler.updateTaskExecutionStateInternal(DefaultScheduler.java:180)
  at org.apache.flink.runtime.scheduler.SchedulerBase.updateTaskExecutionState(SchedulerBase.java:496)
  at org.apache.flink.runtime.scheduler.UpdateSchedulerNgOnInternalFailuresListener.notifyTaskFailure(UpdateSchedulerNgOnInternalFailuresListener.java:49)
  at org.apache.flink.runtime.executiongraph.ExecutionGraph.notifySchedulerNgAboutInternalTaskFailure(ExecutionGraph.java:1703)
  at org.apache.flink.runtime.executiongraph.Execution.processFail(Execution.java:1252)
  at org.apache.flink.runtime.executiongraph.Execution.processFail(Execution.java:1220)
  at org.apache.flink.runtime.executiongraph.Execution.markFailed(Execution.java:1051)
  at org.apache.flink.runtime.executiongraph.ExecutionVertex.markFailed(ExecutionVertex.java:748)
  at org.apache.flink.runtime.scheduler.DefaultExecutionVertexOperations.markFailed(DefaultExecutionVertexOperations.java:41)
  at org.apache.flink.runtime.scheduler.DefaultScheduler.handleTaskDeploymentFailure(DefaultScheduler.java:446)
  at org.apache.flink.runtime.scheduler.DefaultScheduler.lambda$assignResourceOrHandleError$5(DefaultScheduler.java:433)
  at java.util.concurrent.CompletableFuture.uniHandle(CompletableFuture.java:822)
  at java.util.concurrent.CompletableFuture$UniHandle.tryFire(CompletableFuture.java:797)
  at java.util.concurrent.CompletableFuture.postComplete(CompletableFuture.java:474)
  at java.util.concurrent.CompletableFuture.completeExceptionally(CompletableFuture.java:1977)
  at org.apache.flink.runtime.jobmaster.slotpool.SchedulerImpl.lambda$internalAllocateSlot$0(SchedulerImpl.java:168)
  at java.util.concurrent.CompletableFuture.uniWhenComplete(CompletableFuture.java:760)
  at java.util.concurrent.CompletableFuture$UniWhenComplete.tryFire(CompletableFuture.java:736)
  at java.util.concurrent.CompletableFuture.postComplete(CompletableFuture.java:474)
  at java.util.concurrent.CompletableFuture.completeExceptionally(CompletableFuture.java:1977)
  at org.apache.flink.runtime.jobmaster.slotpool.SlotSharingManager$SingleTaskSlot.release(SlotSharingManager.java:726)
  at org.apache.flink.runtime.jobmaster.slotpool.SlotSharingManager$MultiTaskSlot.release(SlotSharingManager.java:537)
  at org.apache.flink.runtime.jobmaster.slotpool.SlotSharingManager$MultiTaskSlot.lambda$new$0(SlotSharingManager.java:432)
  at java.util.concurrent.CompletableFuture.uniHandle(CompletableFuture.java:822)
  at java.util.concurrent.CompletableFuture$UniHandle.tryFire(CompletableFuture.java:797)
  at java.util.concurrent.CompletableFuture.postComplete(CompletableFuture.java:474)
  at java.util.concurrent.CompletableFuture.completeExceptionally(CompletableFuture.java:1977)
  at org.apache.flink.runtime.concurrent.FutureUtils.lambda$forward$21(FutureUtils.java:1065)
  at java.util.concurrent.CompletableFuture.uniWhenComplete(CompletableFuture.java:760)
  at java.util.concurrent.CompletableFuture$UniWhenComplete.tryFire(CompletableFuture.java:736)
  at java.util.concurrent.CompletableFuture.postComplete(CompletableFuture.java:474)
  at java.util.concurrent.CompletableFuture.completeExceptionally(CompletableFuture.java:1977)
  at org.apache.flink.runtime.concurrent.FutureUtils$Timeout.run(FutureUtils.java:999)
  at org.apache.flink.runtime.rpc.akka.AkkaRpcActor.handleRunAsync(AkkaRpcActor.java:402)
  at org.apache.flink.runtime.rpc.akka.AkkaRpcActor.handleRpcMessage(AkkaRpcActor.java:195)
  at org.apache.flink.runtime.rpc.akka.FencedAkkaRpcActor.handleRpcMessage(FencedAkkaRpcActor.java:74)
  at org.apache.flink.runtime.rpc.akka.AkkaRpcActor.handleMessage(AkkaRpcActor.java:152)
  at akka.japi.pf.UnitCaseStatement.apply(CaseStatements.scala:26)
  at akka.japi.pf.UnitCaseStatement.apply(CaseStatements.scala:21)
  at scala.PartialFunction$class.applyOrElse(PartialFunction.scala:123)
  at akka.japi.pf.UnitCaseStatement.applyOrElse(CaseStatements.scala:21)
  at scala.PartialFunction$OrElse.applyOrElse(PartialFunction.scala:170)
  at scala.PartialFunction$OrElse.applyOrElse(PartialFunction.scala:171)
  at scala.PartialFunction$OrElse.applyOrElse(PartialFunction.scala:171)
  at akka.actor.Actor$class.aroundReceive(Actor.scala:517)
  at akka.actor.AbstractActor.aroundReceive(AbstractActor.scala:225)
  at akka.actor.ActorCell.receiveMessage(ActorCell.scala:592)
  at akka.actor.ActorCell.invoke(ActorCell.scala:561)
  at akka.dispatch.Mailbox.processMailbox(Mailbox.scala:258)
  at akka.dispatch.Mailbox.run(Mailbox.scala:225)
  at akka.dispatch.Mailbox.exec(Mailbox.scala:235)
  at akka.dispatch.forkjoin.ForkJoinTask.doExec(ForkJoinTask.java:260)
  at akka.dispatch.forkjoin.ForkJoinPool$WorkQueue.runTask(ForkJoinPool.java:1339)
  at akka.dispatch.forkjoin.ForkJoinPool.runWorker(ForkJoinPool.java:1979)
  at akka.dispatch.forkjoin.ForkJoinWorkerThread.run(ForkJoinWorkerThread.java:107)
Caused by: org.apache.flink.runtime.jobmanager.scheduler.NoResourceAvailableException: Could not allocate the required slot within slot request timeout. Please make sure that the cluster has enough resources.
  at org.apache.flink.runtime.scheduler.DefaultScheduler.maybeWrapWithNoResourceAvailableException(DefaultScheduler.java:452)
  ... 45 more
Caused by: java.util.concurrent.CompletionException: java.util.concurrent.TimeoutException
  at java.util.concurrent.CompletableFuture.encodeThrowable(CompletableFuture.java:292)
  at java.util.concurrent.CompletableFuture.completeThrowable(CompletableFuture.java:308)
  at java.util.concurrent.CompletableFuture.uniApply(CompletableFuture.java:593)
  at java.util.concurrent.CompletableFuture$UniApply.tryFire(CompletableFuture.java:577)
  ... 25 more
Caused by: java.util.concurrent.TimeoutException
  ... 23 more

解决方案是:

$FLINK_HOME/conf/flink-conf.yaml修改如下:

taskmanager.numberOfTaskSlots: 2(这个一般是cpu数量)

parallelism.default: 2

目前感觉是parallelism.default≥taskmanager.numberOfTaskSlots会好些，不然容易报错。

 

Reference:

[1]Apache Flink 零基础入门（四）：客户端操作的 5 种模式

 类似资料：

相关阅读

django model去掉unique_together报错的解决方案node.js缺少mysql模块运行报错的解决方法Nginx中报错：Permission denied与Connection refused的解决php使用pdo连接报错Connection failed SQLSTATE的解决方法vue使用prop可以渲染但是打印台报错的解决方式

相关文章

#中国电信#解决方案经理#综面联想 解决方案产品经理面经七种分布式系统的解决方案，一次性讲给你听！Spring Tx (分布式事务及解决方案)跨域解决方案

相关问答

java - Maven项目构建报错解决方案?instanceof中的模式匹配有时不能解决理论上可解决的范围未处理的警报异常：存在模式对话框（Selenium）Spring云流中的批处理模式和自定义错误处理如何解决 typescript check-types 时的 'originalKeywordKind' 报错?

相关文档

自然语言处理的神经网络模型入门网站设计解构：有效的交互设计框架和模式怎样优化 Pentium 系列处理器的代码学用 JavaScript 设计模式30 秒就能理解的 JavaScript 代码片段
快捷导航： 新手教程 算法原理 架构设计 Java进阶 数据库进阶 大厂专栏 面试经验 编程笔记 编程问答 所有专题 文档资料 工具软件 电子书籍 小牛导航
在线工具： 房贷计算器 个税计算器 Linux命令查询 Json格式化 正则表达式 颜色转换 AES加解密 SHA1加密 MD5加密 毒鸡汤 字数统计 随机密码生成 进制转换 Base64编解码 励志句子
Copyright © 2019-2024 小牛知识库@xnip.cn. All Rights Reserved.