flink-user-zh mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From 戴嘉诚 <a773807...@gmail.com>
Subject 答复: flink集群性能问题
Date Fri, 10 May 2019 09:59:27 GMT
好的,我尝试吧18个flink job拆分出来,在yarn中单独运行,观察里面的jm/tm的信息,谢谢.

发件人: Yun Gao
发送时间: 2019年5月10日 17:52
收件人: user-zh
主题: Re: flink集群性能问题

异常一和异常三应该是相关的,因为Heartbeat也是通过akka来发送的;如果确定不是GC的问题的话,那么超时一般是由于JM/TM在做一些耗时的操作导致akka线程阻塞,但是具体在做什么操作应该还需要进一步分析。


------------------------------------------------------------------
From:戴嘉诚 <a773807943@gmail.com>
Send Time:2019 May 10 (Fri.) 17:00
To:user-zh@flink.apache.org <user-zh@flink.apache.org>
Subject:flink集群性能问题

大家好:
 我这里遇到了一个问题,我的运行方式是flink session on yarn上,一共有18个任务在这个session上运行,这个任务运行了几天后,最近开始发现有几个job,不定时报这个错误,(ps:就这几个job报这个异常,其他job没有出现)。一直都提示超时,然后看了gc,发现没有长时间的的fullgc,而且gc也改为了用g1垃圾收集器,但是也是会有这个问题。
 状态后端使用的是文件后端,以前用rocksDB的时候,也是出现过如此异常。

异常一:
java.util.concurrent.TimeoutException: Heartbeat of TaskManager with id container_e30_1556059873942_0007_01_003306
timed out.
 at org.apache.flink.runtime.jobmaster.JobMaster$TaskManagerHeartbeatListener.notifyHeartbeatTimeout(JobMaster.java:1656)
 at org.apache.flink.runtime.heartbeat.HeartbeatManagerImpl$HeartbeatMonitor.run(HeartbeatManagerImpl.java:339)
 at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511)
 at java.util.concurrent.FutureTask.run(FutureTask.java:266)
 at org.apache.flink.runtime.concurrent.akka.ActorSystemScheduledExecutorAdapter$ScheduledFutureTask.run(ActorSystemScheduledExecutorAdapter.java:154)
 at akka.dispatch.TaskInvocation.run(AbstractDispatcher.scala:39)
 at akka.dispatch.ForkJoinExecutorConfigurator$AkkaForkJoinTask.exec(AbstractDispatcher.scala:415)
 at scala.concurrent.forkjoin.ForkJoinTask.doExec(ForkJoinTask.java:260)
 at scala.concurrent.forkjoin.ForkJoinPool$WorkQueue.runTask(ForkJoinPool.java:1339)
 at scala.concurrent.forkjoin.ForkJoinPool.runWorker(ForkJoinPool.java:1979)
 at scala.concurrent.forkjoin.ForkJoinWorkerThread.run(ForkJoinWorkerThread.java:107)


异常二:
Caused by: org.apache.hadoop.ipc.RemoteException(org.apache.hadoop.hdfs.server.namenode.LeaseExpiredException):
No lease on /flink/checkpoint/public_rule_persona_guanzhu_count/225a05c3ac6192017f38017417ad8f71/chk-6015/1ed16855-3261-4eda-bceb-1519f401eae5
(inode 1967760389): File does not exist. [Lease.  Holder: DFSClient_NONMAPREDUCE_2064130997_92,
pendingcreates: 11]
 at org.apache.hadoop.hdfs.server.namenode.FSNamesystem.checkLease(FSNamesystem.java:3697)
 at org.apache.hadoop.hdfs.server.namenode.FSNamesystem.analyzeFileState(FSNamesystem.java:3498)
 at org.apache.hadoop.hdfs.server.namenode.FSNamesystem.getNewBlockTargets(FSNamesystem.java:3336)
 at org.apache.hadoop.hdfs.server.namenode.FSNamesystem.getAdditionalBlock(FSNamesystem.java:3296)
 at org.apache.hadoop.hdfs.server.namenode.NameNodeRpcServer.addBlock(NameNodeRpcServer.java:850)
 at org.apache.hadoop.hdfs.protocolPB.ClientNamenodeProtocolServerSideTranslatorPB.addBlock(ClientNamenodeProtocolServerSideTranslatorPB.java:504)
 at org.apache.hadoop.hdfs.protocol.proto.ClientNamenodeProtocolProtos$ClientNamenodeProtocol$2.callBlockingMethod(ClientNamenodeProtocolProtos.java)
 at org.apache.hadoop.ipc.ProtobufRpcEngine$Server$ProtoBufRpcInvoker.call(ProtobufRpcEngine.java:640)
 at org.apache.hadoop.ipc.RPC$Server.call(RPC.java:982)
 at org.apache.hadoop.ipc.Server$Handler$1.run(Server.java:2351)
 at org.apache.hadoop.ipc.Server$Handler$1.run(Server.java:2347)
 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:1869)
 at org.apache.hadoop.ipc.Server$Handler.run(Server.java:2347)

 at org.apache.hadoop.ipc.Client.getRpcResponse(Client.java:1554)
 at org.apache.hadoop.ipc.Client.call(Client.java:1498)
 at org.apache.hadoop.ipc.Client.call(Client.java:1398)
 at org.apache.hadoop.ipc.ProtobufRpcEngine$Invoker.invoke(ProtobufRpcEngine.java:233)
 at com.sun.proxy.$Proxy17.addBlock(Unknown Source)
 at org.apache.hadoop.hdfs.protocolPB.ClientNamenodeProtocolTranslatorPB.addBlock(ClientNamenodeProtocolTranslatorPB.java:459)
 at sun.reflect.GeneratedMethodAccessor51.invoke(Unknown Source)
 at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
 at java.lang.reflect.Method.invoke(Method.java:498)
 at org.apache.hadoop.io.retry.RetryInvocationHandler.invokeMethod(RetryInvocationHandler.java:290)
 at org.apache.hadoop.io.retry.RetryInvocationHandler.invoke(RetryInvocationHandler.java:202)
 at org.apache.hadoop.io.retry.RetryInvocationHandler.invoke(RetryInvocationHandler.java:184)
 at com.sun.proxy.$Proxy18.addBlock(Unknown Source)
 at org.apache.hadoop.hdfs.DFSOutputStream$DataStreamer.locateFollowingBlock(DFSOutputStream.java:1580)
 at org.apache.hadoop.hdfs.DFSOutputStream$DataStreamer.nextBlockOutputStream(DFSOutputStream.java:1375)
 at org.apache.hadoop.hdfs.DFSOutputStream$DataStreamer.run(DFSOutputStream.java:564)

异常三:
java.util.concurrent.CompletionException: akka.pattern.AskTimeoutException: Ask timed out
on [Actor[akka.tcp://flink@ambari.host08.yy:37375/user/taskmanager_0#-433497337]] after [10000
ms]. Sender[null] sent message of type "org.apache.flink.runtime.rpc.messages.RemoteRpcInvocation".
 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)
 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$1.onComplete(FutureUtils.java:815)
 at akka.dispatch.OnComplete.internal(Future.scala:258)
 at akka.dispatch.OnComplete.internal(Future.scala:256)
 at akka.dispatch.japi$CallbackBridge.apply(Future.scala:186)
 at akka.dispatch.japi$CallbackBridge.apply(Future.scala:183)
 at scala.concurrent.impl.CallbackRunnable.run(Promise.scala:36)
 at org.apache.flink.runtime.concurrent.Executors$DirectExecutionContext.execute(Executors.java:74)
 at scala.concurrent.impl.CallbackRunnable.executeWithValue(Promise.scala:44)
 at scala.concurrent.impl.Promise$DefaultPromise.tryComplete(Promise.scala:252)
 at akka.pattern.PromiseActorRef$$anonfun$1.apply$mcV$sp(AskSupport.scala:603)
 at akka.actor.Scheduler$$anon$4.run(Scheduler.scala:126)
 at scala.concurrent.Future$InternalCallbackExecutor$.unbatchedExecute(Future.scala:601)
 at scala.concurrent.BatchingExecutor$class.execute(BatchingExecutor.scala:109)
 at scala.concurrent.Future$InternalCallbackExecutor$.execute(Future.scala:599)
 at akka.actor.LightArrayRevolverScheduler$TaskHolder.executeTask(LightArrayRevolverScheduler.scala:329)
 at akka.actor.LightArrayRevolverScheduler$$anon$4.executeBucket$1(LightArrayRevolverScheduler.scala:280)
 at akka.actor.LightArrayRevolverScheduler$$anon$4.nextTick(LightArrayRevolverScheduler.scala:284)
 at akka.actor.LightArrayRevolverScheduler$$anon$4.run(LightArrayRevolverScheduler.scala:236)
 at java.lang.Thread.run(Thread.java:748)
Caused by: akka.pattern.AskTimeoutException: Ask timed out on [Actor[akka.tcp://flink@ambari.host08.yy:37375/user/taskmanager_0#-433497337]]
after [10000 ms]. Sender[null] sent message of type "org.apache.flink.runtime.rpc.messages.RemoteRpcInvocation".
 at akka.pattern.PromiseActorRef$$anonfun$1.apply$mcV$sp(AskSupport.scala:604)
 ... 9 more


其中,这里异常二中,的hdfs文件我本地或者程序都没有操作过,就只有这个job来操作checkpoint,但是,也有时候会报这个异常,我感觉这里三个异常,都是跟一个原因相关,但是具体是什么原因,我这边还是排查不出来,请各位指点一下。

Mime
  • Unnamed multipart/alternative (inline, None, 0 bytes)
View raw message