flink1.10三节点集群standalone模式搭建

最新推荐文章于 2023-04-26 15:23:43 发布

原创最新推荐文章于 2023-04-26 15:23:43 发布

· 3.4k 阅读

1 ·

版权

flink 专栏收录该内容

18 篇文章

订阅专栏

各台机器上提前准备jdk1.8以及上的java环境，并且配置ssh免密登录。

集群环境

flink1:172.21.89.128	jobmanager
flink2:172.21.89.129	taskmanager
flink3:172.21.89.130	taskmanager

在flink1上做flink配置，主要是flink-conf.yaml、masters和slaves

flink-conf.yaml：

jobmanager.rpc.address: flink1
# 每个taskmanager机器提供的slot数量
taskmanager.numberOfTaskSlots: 2
# 默认并行度
parallelism.default: 2
# 临时文件存储路径。需要提前创建，否则启动集群会报错。
io.tmp.dirs: /root/flink/tmp

slaves：

flink2

配置完之后通过scp -r ./flink-1.10.0 flink2:/opt将flink文件传输到flink2。之后就可以在flink1上启动集群：bin/start-cluster.sh

我这里因为只在flink1上提前创建了临时文件目录/root/flink/tmp，而flink2上没有，所以启动时，jobmanager启动成功，但taskmanager启动失败：

2020-05-19 16:03:32,276 INFO  org.apache.flink.runtime.security.modules.HadoopModuleFactory  - Cannot create Hadoop Security Module because Hadoop cannot be found in the Classpath.
2020-05-19 16:03:32,645 ERROR org.apache.flink.runtime.taskexecutor.TaskManagerRunner       - TaskManager initialization failed.
java.lang.Exception: unable to establish the security context
        at org.apache.flink.runtime.security.SecurityUtils.install(SecurityUtils.java:73)
        at org.apache.flink.runtime.taskexecutor.TaskManagerRunner.runTaskManagerSecurely(TaskManagerRunner.java:319)
        at org.apache.flink.runtime.taskexecutor.TaskManagerRunner.main(TaskManagerRunner.java:287)
Caused by: java.lang.RuntimeException: unable to generate a JAAS configuration file
        at org.apache.flink.runtime.security.modules.JaasModule.generateDefaultConfigFile(JaasModule.java:170)
        at org.apache.flink.runtime.security.modules.JaasModule.install(JaasModule.java:94)
        at org.apache.flink.runtime.security.SecurityUtils.install(SecurityUtils.java:67)
        ... 2 more
Caused by: java.nio.file.NoSuchFileException: /root/flink/tmp/jaas-7614411253117836328.conf
        at sun.nio.fs.UnixException.translateToIOException(UnixException.java:86)
        at sun.nio.fs.UnixException.rethrowAsIOException(UnixException.java:102)
        at sun.nio.fs.UnixException.rethrowAsIOException(UnixException.java:107)
        at sun.nio.fs.UnixFileSystemProvider.newByteChannel(UnixFileSystemProvider.java:214)
        at java.nio.file.Files.newByteChannel(Files.java:361)
        at java.nio.file.Files.createFile(Files.java:632)
        at java.nio.file.TempFileHelper.create(TempFileHelper.java:138)
        at java.nio.file.TempFileHelper.createTempFile(TempFileHelper.java:161)
        at java.nio.file.Files.createTempFile(Files.java:852)
        at org.apache.flink.runtime.security.modules.JaasModule.generateDefaultConfigFile(JaasModule.java:163)
        ... 4 more

刚好，这样可以测试动态添加taskmanager。首先在flink2上创建临时文件目录/root/flink/tmp，然后在flink2上运行./taskmanager.sh start实现动态添加taskmanager节点。

启动后看taskmanager日志：

此时可以在jobmanager的日志上看到此taskmanager注册成功的日志，其中ResourceID与taskmanager日志中的一致：

2020-05-19 08:17:54,099 INFO  org.apache.flink.runtime.resourcemanager.StandaloneResourceManager  - Registering TaskManager with ResourceID f6fad8c42b36967dabf400ffb52da4df (akka.tcp
://flink@172.21.89.129:44783/user/taskmanager_0) at ResourceManager

对于flink3，也采用动态添加的方式（没有在slaves配置文件中声明flink3）。首先在flink3上创建临时文件目录/root/flink/tmp，然后在flink3上运行./taskmanager.sh start实现动态添加taskmanager节点。

此时可以在jobmanager的日志上看到此taskmanager注册成功的日志，其中ResourceID与taskmanager日志中的一致：

2020-05-19 08:17:54,099 INFO  org.apache.flink.runtime.resourcemanager.StandaloneResourceManager  - Registering TaskManager with ResourceID f6fad8c42b36967dabf400ffb52da4df (akka.tcp://flink@172.21.89.129:44783/user/taskmanager_0) at ResourceManager
2020-05-19 08:46:03,676 INFO  org.apache.flink.runtime.resourcemanager.StandaloneResourceManager  - Registering TaskManager with ResourceID e008aba19fdb81984b393f19a24df5f1 (akka.tcp://flink@172.21.89.130:36784/user/taskmanager_0) at ResourceManager

webui可以看到taskmanager和jobmanager的配置和使用情况

总结：

本文介绍了standalone模式的非高可用的三节点集群搭建
介绍了动态添加taskmanager的过程，需要注意的是，动态添加的机器因为没有在slaves文件中配置，因此在使用stop-cluster.sh脚本关闭flink集群的时候，会无法正常关闭此动态添加的机器（下面的日志是执行stop-cluster.sh脚本后，flink3中的taskmanager日志，可以看到flink3上的taskmanager尝试与flink1的resourceManager通信失败，最终关闭taskmanager）。

2020-05-19 16:43:42,100 INFO  org.apache.flink.runtime.taskexecutor.TaskExecutor            - Successful registration at resource manager akka.tcp://flink@flink
1:6123/user/resourcemanager under registration id e09e4d6b3c15e36d26f7f6569cdecd76.
2020-05-19 18:05:56,030 WARN  akka.remote.ReliableDeliverySupervisor                        - Association with remote system [akka.tcp://flink@flink1:6123] has
failed, address is now gated for [50] ms. Reason: [Disassociated]
2020-05-19 18:11:25,751 INFO  org.apache.flink.runtime.taskexecutor.TaskExecutor            - Could not resolve ResourceManager address akka.tcp://flink@flink1:6123/user/resourcemanager, retrying in 10000 ms: Could not connect to rpc endpoint under address akka.tcp://flink@flink1:6123/user/resourcemanager..
2020-05-19 18:11:35,769 WARN  akka.remote.transport.netty.NettyTransport                    - Remote connection to [null] failed with java.net.ConnectException: 拒绝连接: flink1/172.21.89.128:6123
2020-05-19 18:11:35,770 WARN  akka.remote.ReliableDeliverySupervisor                        - Association with remote system [akka.tcp://flink@flink1:6123] has failed, address is now gated for [50] ms. Reason: [Association failed with [akka.tcp://flink@flink1:6123]] Caused by: [java.net.ConnectException: 拒绝连接: flink1/172.21.89.128:6123]
2020-05-19 18:11:35,771 INFO  org.apache.flink.runtime.taskexecutor.TaskExecutor            - Could not resolve ResourceManager address akka.tcp://flink@flink1:6123/user/resourcemanager, retrying in 10000 ms: Could not connect to rpc endpoint under address akka.tcp://flink@flink1:6123/user/resourcemanager..
2020-05-19 18:11:45,086 ERROR org.apache.flink.runtime.taskexecutor.TaskExecutor            - Fatal error occurred in TaskExecutor akka.tcp://flink@172.21.89.130:36784/user/taskmanager_0.
org.apache.flink.runtime.taskexecutor.exceptions.RegistrationTimeoutException: Could not register at the ResourceManager within the specified maximum registration duration 300000 ms. This indicates a problem with this instance. Terminating now.
        at org.apache.flink.runtime.taskexecutor.TaskExecutor.registrationTimeout(TaskExecutor.java:1120)
        at org.apache.flink.runtime.taskexecutor.TaskExecutor.lambda$startRegistrationTimeout$9(TaskExecutor.java:1106)
        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.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.applyOrElse(PartialFunction.scala:123)
        at scala.PartialFunction.applyOrElse$(PartialFunction.scala:122)
        at akka.japi.pf.UnitCaseStatement.applyOrElse(CaseStatements.scala:21)
        at scala.PartialFunction$OrElse.applyOrElse(PartialFunction.scala:171)
        at scala.PartialFunction$OrElse.applyOrElse(PartialFunction.scala:172)
        at scala.PartialFunction$OrElse.applyOrElse(PartialFunction.scala:172)
        at akka.actor.Actor.aroundReceive(Actor.scala:517)
        at akka.actor.Actor.aroundReceive$(Actor.scala:515)
        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)
2020-05-19 18:11:45,088 ERROR org.apache.flink.runtime.taskexecutor.TaskManagerRunner       - Fatal error occurred while executing the TaskManager. Shutting it down...
org.apache.flink.runtime.taskexecutor.exceptions.RegistrationTimeoutException: Could not register at the ResourceManager within the specified maximum registration duration 300000 ms. This indicates a problem with this instance. Terminating now.
        at org.apache.flink.runtime.taskexecutor.TaskExecutor.registrationTimeout(TaskExecutor.java:1120)
        at org.apache.flink.runtime.taskexecutor.TaskExecutor.lambda$startRegistrationTimeout$9(TaskExecutor.java:1106)
        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.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.applyOrElse(PartialFunction.scala:123)
        at scala.PartialFunction.applyOrElse$(PartialFunction.scala:122)
        at akka.japi.pf.UnitCaseStatement.applyOrElse(CaseStatements.scala:21)
        at scala.PartialFunction$OrElse.applyOrElse(PartialFunction.scala:171)
        at scala.PartialFunction$OrElse.applyOrElse(PartialFunction.scala:172)
        at scala.PartialFunction$OrElse.applyOrElse(PartialFunction.scala:172)
        at akka.actor.Actor.aroundReceive(Actor.scala:517)
        at akka.actor.Actor.aroundReceive$(Actor.scala:515)
        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)
2020-05-19 18:11:45,093 INFO  org.apache.flink.runtime.taskexecutor.TaskExecutor            - Stopping TaskExecutor akka.tcp://flink@172.21.89.130:36784/user/taskmanager_0.
2020-05-19 18:11:45,094 INFO  org.apache.flink.runtime.taskexecutor.TaskExecutor            - Terminating registration attempts towards ResourceManager akka.tcp://flink@flink1:6123/user/resourcemanager.
2020-05-19 18:11:45,097 INFO  org.apache.flink.runtime.taskexecutor.JobLeaderService        - Stop job leader service.
2020-05-19 18:11:45,097 INFO  org.apache.flink.runtime.state.TaskExecutorLocalStateStoresManager  - Shutting down TaskExecutorLocalStateStoresManager.
2020-05-19 18:11:45,108 INFO  org.apache.flink.runtime.io.disk.FileChannelManagerImpl       - FileChannelManager removed spill file directory /root/flink/tmp/flink-io-e748c002-6239-47e7-9dea-2901689d750b
2020-05-19 18:11:45,108 INFO  org.apache.flink.runtime.io.network.NettyShuffleEnvironment   - Shutting down the network environment and its components.
2020-05-19 18:11:45,109 INFO  org.apache.flink.runtime.io.network.netty.NettyClient         - Successful shutdown (took 1 ms).
2020-05-19 18:11:45,111 INFO  org.apache.flink.runtime.io.network.netty.NettyServer         - Successful shutdown (took 2 ms).
2020-05-19 18:11:45,113 INFO  org.apache.flink.runtime.io.disk.FileChannelManagerImpl       - FileChannelManager removed spill file directory /root/flink/tmp/flink-netty-shuffle-57c232aa-407c-4a78-9eaa-5d2a233b2f06
2020-05-19 18:11:45,114 INFO  org.apache.flink.runtime.taskexecutor.KvStateService          - Shutting down the kvState service and its components.
2020-05-19 18:11:45,114 INFO  org.apache.flink.runtime.taskexecutor.JobLeaderService        - Stop job leader service.
2020-05-19 18:11:45,114 INFO  org.apache.flink.runtime.filecache.FileCache                  - removed file cache directory /root/flink/tmp/flink-dist-cache-665f8fb6-d15f-4fea-8d63-7e9fe9031f5f
2020-05-19 18:11:45,117 INFO  org.apache.flink.runtime.taskexecutor.TaskExecutor            - Stopped TaskExecutor akka.tcp://flink@172.21.89.130:36784/user/taskmanager_0.
2020-05-19 18:11:45,117 INFO  org.apache.flink.runtime.blob.PermanentBlobCache              - Shutting down BLOB cache
2020-05-19 18:11:45,117 INFO  org.apache.flink.runtime.blob.TransientBlobCache              - Shutting down BLOB cache
2020-05-19 18:11:45,117 INFO  org.apache.flink.runtime.rpc.akka.AkkaRpcService              - Stopping Akka RPC service.
2020-05-19 18:11:45,134 INFO  org.apache.flink.runtime.rpc.akka.AkkaRpcService              - Stopping Akka RPC service.
2020-05-19 18:11:45,138 INFO  akka.remote.RemoteActorRefProvider$RemotingTerminator         - Shutting down remote daemon.
2020-05-19 18:11:45,164 INFO  akka.remote.RemoteActorRefProvider$RemotingTerminator         - Remote daemon shut down; proceeding with flushing remote transports.
2020-05-19 18:11:45,140 INFO  akka.remote.RemoteActorRefProvider$RemotingTerminator         - Shutting down remote daemon.
2020-05-19 18:11:45,222 INFO  akka.remote.RemoteActorRefProvider$RemotingTerminator         - Remote daemon shut down; proceeding with flushing remote transports.
2020-05-19 18:11:45,270 INFO  akka.remote.RemoteActorRefProvider$RemotingTerminator         - Remoting shut down.
2020-05-19 18:11:45,277 INFO  akka.remote.RemoteActorRefProvider$RemotingTerminator         - Remoting shut down.
2020-05-19 18:11:45,318 INFO  org.apache.flink.runtime.rpc.akka.AkkaRpcService              - Stopped Akka RPC service.