logo标签列表

Flink任务管理器在无法完成检查点/保存点后陷入停顿（CPU使用率为100%）

apache-flinkflink-streamingamazon-kinesis
3

-- 通过将状态后端从文件系统更改为rocksdb解决了问题 --

在AWS EMR上运行Flink 1.9。 Flink应用程序使用kinesis流作为输入数据，另一个kinesis流作为输出。 最近，检查点大小已经增长到1千兆字节（由于更多数据）。 有时，在尝试进行检查点时 - 应用程序开始利用整个处理器资源（每天发生几次）

指标：

LA（带有作业/任务管理器的emr ec2核心节点）

Run Loop Time - kinesis消费者

Records Per Fetch - kinesis消费者

Task manager GC

jobmanager日志

{"level":"INFO","timestamp":"2020-08-25 04:55:27,399","thread":"Checkpoint Timer","file":"CheckpointCoordinator.java","line":"617","message":"Triggering checkpoint 1232 @ 1598331327244 for job 0039825bafae26bc34db88e037a1dae3."}

{"level":"INFO","timestamp":"2020-08-25 04:58:24,509","thread":"flink-akka.actor.default-dispatcher-7010","file":"ResourceManager.java","line":"1144","message":"The heartbeat of TaskManager with id container_1597960565773_0003_01_000002 timed out."}

{"level":"INFO","timestamp":"2020-08-25 04:58:24,510","thread":"flink-akka.actor.default-dispatcher-7010","file":"ResourceManager.java","line":"805","message":"Closing TaskExecutor connection container_1597960565773_0003_01_000002 because: The heartbeat of TaskManager with id container_1597960565773_0003_01_000002  timed out."}
{"level":"INFO","timestamp":"2020-08-25 04:58:24,514","thread":"flink-akka.actor.default-dispatcher-7015","file":"Execution.java","line":"1493","message":"Sink: kinesis-events-sink (1/1) (573401e241fe0a0ac0a8a54c81c4eefd) switched from RUNNING to FAILED."}
java.util.concurrent.TimeoutException: The heartbeat of TaskManager with id container_1597960565773_0003_01_000002  timed out.
        at org.apache.flink.runtime.resourcemanager.ResourceManager$TaskManagerHeartbeatListener.notifyHeartbeatTimeout(ResourceManager.java:1146)
        at org.apache.flink.runtime.heartbeat.HeartbeatMonitorImpl.run(HeartbeatMonitorImpl.java:109)
        at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511)
        at java.util.concurrent.FutureTask.run(FutureTask.java:266)
        at org.apache.flink.runtime.rpc.akka.AkkaRpcActor.handleRunAsync(AkkaRpcActor.java:397)
        at org.apache.flink.runtime.rpc.akka.AkkaRpcActor.handleRpcMessage(AkkaRpcActor.java:190)
        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)
{"level":"INFO","timestamp":"2020-08-25 04:58:24,515","thread":"flink-akka.actor.default-dispatcher-7015","file":"ExecutionGraph.java","line":"1324","message":"Job JOB_NAME_HIDDEN (0039825bafae26bc34db88e037a1dae3) switched from state RUNNING to FAILING."}
java.util.concurrent.TimeoutException: The heartbeat of TaskManager with id container_1597960565773_0003_01_000002  timed out.
        at org.apache.flink.runtime.resourcemanager.ResourceManager$TaskManagerHeartbeatListener.notifyHeartbeatTimeout(ResourceManager.java:1146)
        at org.apache.flink.runtime.heartbeat.HeartbeatMonitorImpl.run(HeartbeatMonitorImpl.java:109)
        at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511)
        at java.util.concurrent.FutureTask.run(FutureTask.java:266)
        at org.apache.flink.runtime.rpc.akka.AkkaRpcActor.handleRunAsync(AkkaRpcActor.java:397)
        at org.apache.flink.runtime.rpc.akka.AkkaRpcActor.handleRpcMessage(AkkaRpcActor.java:190)
        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)

任务管理器日志

{"level":"INFO","timestamp":"2020-08-25 04:58:17,504","thread":"kpl-daemon-0003","file":"LogInputStreamReader.java","line":"59","message":"[2020-08-25 04:58:02.469832] [0x00007394][0x00007f2673d70700] [info] [processing_statistics_logger.cc:127] (sensing-events-test) Aver
age Processing Time: -nan ms"}
{"level":"INFO","timestamp":"2020-08-25 04:58:17,504","thread":"kpl-daemon-0003","file":"LogInputStreamReader.java","line":"59","message":"[2020-08-25 04:58:17.469922] [0x00007394][0x00007f2673d70700] [info] [processing_statistics_logger.cc:109] Stage 1 Triggers: { stream
: 'sensing-events-test', manual: 0, count: 0, size: 0, matches: 0, timed: 0, UserRecords: 0, KinesisRecords: 0 }"}
{"level":"INFO","timestamp":"2020-08-25 04:58:17,504","thread":"kpl-daemon-0003","file":"LogInputStreamReader.java","line":"59","message":"[2020-08-25 04:58:17.469977] [0x00007394][0x00007f2673d70700] [info] [processing_statistics_logger.cc:112] Stage 2 Triggers: { stream
: 'sensing-events-test', manual: 0, count: 0, size: 0, matches: 0, timed: 0, KinesisRecords: 0, PutRecords: 0 }"}
{"level":"INFO","timestamp":"2020-08-25 04:58:17,504","thread":"kpl-daemon-0003","file":"LogInputStreamReader.java","line":"59","message":"[2020-08-25 04:58:17.469992] [0x00007394][0x00007f2673d70700] [info] [processing_statistics_logger.cc:127] (sensing-events-test) Aver
age Processing Time: -nan ms"}
{"level":"ERROR","timestamp":"2020-08-25 04:58:28,535","thread":"flink-akka.actor.default-dispatcher-628","file":"FatalExitExceptionHandler.java","line":"40","message":"FATAL: Thread 'flink-akka.actor.default-dispatcher-628' produced an uncaught exception. Stopping the pr
ocess..."}
java.lang.OutOfMemoryError: GC overhead limit exceeded
        at sun.reflect.UTF8.encode(UTF8.java:36)
        at sun.reflect.ClassFileAssembler.emitConstantPoolUTF8(ClassFileAssembler.java:103)
        at sun.reflect.MethodAccessorGenerator.generate(MethodAccessorGenerator.java:331)
        at sun.reflect.MethodAccessorGenerator.generateSerializationConstructor(MethodAccessorGenerator.java:112)
        at sun.reflect.ReflectionFactory.generateConstructor(ReflectionFactory.java:398)
        at sun.reflect.ReflectionFactory.newConstructorForSerialization(ReflectionFactory.java:360)
        at java.io.ObjectStreamClass.getSerializableConstructor(ObjectStreamClass.java:1588)
        at java.io.ObjectStreamClass.access$1500(ObjectStreamClass.java:79)
        at java.io.ObjectStreamClass$3.run(ObjectStreamClass.java:519)
        at java.io.ObjectStreamClass$3.run(ObjectStreamClass.java:494)
        at java.security.AccessController.doPrivileged(Native Method)
        at java.io.ObjectStreamClass.<init>(ObjectStreamClass.java:494)
        at java.io.ObjectStreamClass.lookup(ObjectStreamClass.java:391)
        at java.io.ObjectStreamClass.initNonProxy(ObjectStreamClass.java:681)
        at java.io.ObjectInputStream.readNonProxyDesc(ObjectInputStream.java:1941)
        at java.io.ObjectInputStream.readClassDesc(ObjectInputStream.java:1807)
        at java.io.ObjectInputStream.readClass(ObjectInputStream.java:1770)
        at java.io.ObjectInputStream.readObject0(ObjectInputStream.java:1595)
        at java.io.ObjectInputStream.readObject(ObjectInputStream.java:464)
        at java.io.ObjectInputStream.readObject(ObjectInputStream.java:422)
        at org.apache.flink.runtime.rpc.messages.RemoteRpcInvocation$MethodInvocation.readObject(RemoteRpcInvocation.java:204)
        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 java.io.ObjectStreamClass.invokeReadObject(ObjectStreamClass.java:1184)
        at java.io.ObjectInputStream.readSerialData(ObjectInputStream.java:2234)
        at java.io.ObjectInputStream.readOrdinaryObject(ObjectInputStream.java:2125)
        at java.io.ObjectInputStream.readObject0(ObjectInputStream.java:1624)
        at java.io.ObjectInputStream.readObject(ObjectInputStream.java:464)
        at java.io.ObjectInputStream.readObject(ObjectInputStream.java:422)
        at org.apache.flink.util.InstantiationUtil.deserializeObject(InstantiationUtil.java:576)

upd. flink-conf.yaml

state.backend.fs.checkpointdir: s3a://s3-bucket-with-checkpoints/flink-checkpoints
taskmanager.numberOfTaskSlots: 1
state.backend: filesystem
taskmanager.heap.size: 3057m
state.checkpoints.dir: s3a://s3-bucket-with-checkpoints/external-checkpoints

Flink检查点

2个回答
1
我认为，这可能与SlidingEventTimeWindow有关。根据检查点截图，我了解到它是一个大小为2分钟、窗口滑动为2秒的窗口。Flink会为每个元素创建属于其所属窗口的副本。因此，在您的情况下，对于滑动窗口，它会创建大约60个元素副本，因此状态大小比滚动窗口大60倍。
我猜，在检查点时，Flink尝试序列化状态，但内存不足，因此启动了GC，并最终导致内存耗尽。
从图表来看，当Flink开始制作保存点时，Kinesis客户端停止读取，然后在检查点之后，Flink会获得“冲击”负载。 这是应该的吗？ 我需要调整哪个内存参数才能拥有足够的内存？在图表上，我没有看到内存耗尽。并且这种情况在相同的负载下偶然发生。 - Sergey Postument
同样实现CheckpointedFunction接口的源必须确保状态检查点、内部状态更新和元素发射不会同时进行。这可以通过使用检查点锁对象在同步块中保护状态更新和元素发射来实现。因此，新记录只能在同步块内收集，因此当快照开始时，消费者等待快照完成。从内存角度来看，每个TM上的flink管理内存量是多少？您能提供您的flink-conf.yaml设置吗？谢谢。 - Mikalai Lushchytski
非常抱歉问了这么多问题，但是您能否进一步澄清您是否在独立模式下启动flink集群或任何容器化环境上？此外，您使用的flink版本是什么？自1.10以来，内存管理已经发生了变化。通常，在文件系统状态后端的情况下，建议将管理内存设置为零。这将确保在JVM上为用户代码分配最大的内存量。 - Mikalai Lushchytski
Flink 1.9 在 AWS EMR（Yarn）上运行。 如何禁用托管内存？也许我需要将状态后端从文件系统更改为其他后端？ - Sergey Postument
2
通过将状态后端从文件系统更改为RocksDB，解决了问题。 - Sergey Postument
显示剩余2条评论
1

通过将状态后端从文件系统更改为rocksdb解决了问题。

网页内容由stack overflow 提供, 点击上面的
可以查看英文原文，
原文链接