首页 > 其他分享 >搭建Flink集群、集群HA高可用以及配置历史服务器

搭建Flink集群、集群HA高可用以及配置历史服务器

时间：2023-10-15 17:07:08浏览次数：39  
标签：node02 flink node01 Flink host 集群 HA root

Flink集群搭建

集群规划

节点 node01 node02 node03
角色 JobManager <br> TaskManager TaskManager TaskManager

下载并解压安装包

wget https://repo.huaweicloud.com/apache/flink/flink-1.17.0/flink-1.17.0-bin-scala_2.12.tgz

在node01节点下载flink安装包，同时解压、重命名。

tar  -zxvf flink-1.17.0-bin-scala_2.12.tgz 

mv flink-1.17.0 flink

修改集群配置

进入flink的conf目录，修改集群配置

vim /usr/local/program/flink/conf/flink-conf.yaml

1.修改flink-conf.yaml文件

JobManager节点配置

# jobmanager.rpc.address: localhost
# jobmanager.bind-host: localhost
jobmanager.rpc.address: node01
jobmanager.bind-host: 0.0.0.0

# rest.address: localhost
# rest.bind-address: localhost
rest.address: node01
rest.bind-address: 0.0.0.0

TaskManager节点配置

# taskmanager.host: localhost
# taskmanager.bind-host: localhost

taskmanager.host: node01
taskmanager.bind-host: 0.0.0.0

注意：需要在/etc/hosts文件中配置各个节点信息

172.29.234.1	node01	node01
172.29.234.2	node02	node02
172.29.234.3	node03	node03

2.修改workers文件

指定node01、node02、node03等节点为TaskManager

# localhost
node01
node02
node03

3.修改masters文件

# localhost:8081
node01:8081

分发安装目录

在node01节点安装、配置好后，将Flink安装目录分发给另外两个节点服务器。

[root@node01 program]# pwd
/usr/local/program
[root@node01 program]# ls
flink                            jdk8

[root@node01 program]# scp -r flink node02:/usr/local/program/flink

[root@node01 program]# scp -r flink node03:/usr/local/program/flink

在node02、node03节点，修改flink-conf.yaml配置

1.node02节点

# taskmanager.host: localhost

taskmanager.host: node02

2.node03节点

# taskmanager.host: localhost

taskmanager.host: node03

启动集群

Flink附带了相关的bash脚本，可以用于启动、停止集群。

# 启动集群
./bin/start-cluster.sh

# 停止集群
./bin/stop-cluster.sh

在node01节点服务器上执行start-cluster.sh脚本以启动Flink集群

[root@node01 bin]# cd /usr/local/program/flink/bin

[root@node01 bin]# ./start-cluster.sh 
Starting cluster.
Starting standalonesession daemon on host node01.
Starting taskexecutor daemon on host node01.
Starting taskexecutor daemon on host node02.
Starting taskexecutor daemon on host node03.

查看进程情况

[root@node01 bin]# jps
6788 StandaloneSessionClusterEntrypoint
7256 Jps
7116 TaskManagerRunner

[root@node02 conf]# jps
16884 TaskManagerRunner
16959 Jps

[root@node03 conf]# jps
17139 TaskManagerRunner
17214 Jps

访问Web UI

当如上所示一样后，代表启动成功，此时可以访问http://node01:8081对flink集群和任务进行监控管理。

在这里插入图片描述 注意：关闭防火墙，否则可能无法访问，或者集群的TaskManager数量、Slot数量显示异常

systemctl stop firewalld

Flink集群HA高可用

概述

集群实际上只有一个JobManager，是存在单点故障的，官方提供了Standalone Cluster HA模式来实现集群高可用。

集群可以有多个JobManager，但只有一个处于active状态，其余的则处于备用状态，Flink使用 ZooKeeper来选举出Active JobManager，并依赖其来提供一致性协调服务，所以需要预先安装 ZooKeeper 。

Flink本身提供了内置ZooKeeper插件，可以直接修改conf/zoo.cfg，并且使用/bin/start-zookeeper-quorum.sh直接启动。

集群规划

节点 node01 node02 node03
角色 JobManager <br> TaskManager JobManager <br>TaskManager TaskManager

配置flink

基于Flink集群的node01节点配置的情况下，修改conf/flink-conf.yaml文件，增加如下配置：

# 配置使用zookeeper来开启高可用模式
high-availability.type: zookeeper

# 配置zookeeper的地址，采用zookeeper集群时，可以使用逗号来分隔多个节点地址
high-availability.zookeeper.quorum: node01:2181,node02:2181,node03:2181

# 在zookeeper上存储flink集群元信息的路径
high-availability.zookeeper.path.root: /flink

# 集群id 放置集群的所有必需协调数据
high-availability.cluster-id: /cluster_one

# 持久化存储JobManager元数据的地址，zookeeper上存储的只是指向该元数据的指针信息
high-availability.storageDir: hdfs://node01:9000/flink/recovery

配置master、workers

修改conf/masters文件，配置master节点

node01:8081
node02:8081

修改conf/workers文件，配置worker节点

node01
node02
node03

配置ZK

编辑vim zoo.cfg文件

server.1=node01:2888:3888
server.2=node02:2888:3888
server.3=node03:2888:3888

分发安装目录

在node01节点安装、配置好后，将Flink安装目录分发给另外两个节点服务器。

[root@node01 program]# pwd
/usr/local/program
[root@node01 program]# ls
flink                            jdk8

[root@node01 program]# scp -r flink node02:/usr/local/program/flink

[root@node01 program]# scp -r flink node03:/usr/local/program/flink

在node02、node03节点，修改flink-conf.yaml配置

1.node02节点

jobmanager.rpc.address: node02

taskmanager.host: node02

2.node03节点

taskmanager.host: node03

启动HA集群

分发Flink相关配置到其他节点，然后确保Hadoop和ZooKeeper已经启动后，使用以下命令来启动集群：

[root@node01 flink]# bin/start-cluster.sh
Starting HA cluster with 2 masters.
Starting standalonesession daemon on host node01.
Starting standalonesession daemon on host node02.
Starting taskexecutor daemon on host node01.
Starting taskexecutor daemon on host node02.
Starting taskexecutor daemon on host node03.

访问http://node01:8081 在这里插入图片描述 访问http://node02:8081 在这里插入图片描述

测试

查看ZK：JobManager节点信息 在这里插入图片描述 kill node01节点上的JobManager进程

[root@node01 flink]# jps
2564 DataNode
3508 NodeManager
18741 Jps
7784 QuorumPeerMain
16666 TaskManagerRunner
2363 NameNode
16300 StandaloneSessionClusterEntrypoint
3117 ResourceManager
[root@node01 flink]# kill -9 16300

查看Active JobManager是否变化 在这里插入图片描述

Flink参数配置

flink-conf.yaml文件中有大量的配置参数，基本常见参数如下：

# jobmanager地址	
jobmanager.rpc.address: node01

# JobManager 的 JVM 堆内存大小，默认为 1024m 
jobmanager.heap.size: 1024m

# rpc通信端口
jobmanager.rpc.port: 6123

# 进程使用的全部内存大小,可以根据集群规模进行适当调整
jobmanager.memory.process.size：1600m

# Taskmanager 的 JVM 堆内存大小，默认为 1024m 
taskmanager.heap.size: 1024m

# 进程使用的全部内存大小,可以根据集群规模进行适当调整
taskmanager.memory.process.size: 1728m

# 每个TaskManager能够分配的Slot数量进行配置，默认为1 
# 通常设置为 CPU 核心的数量，或其一半
# Slot就是TaskManager中具体运行一个任务所分配的计算资源
taskmanager.numberOfTaskSlots: 1

# flink任务执行的并行度，默认为1
# 优先级低于代码中进行的并行度配置和任务提交时使用参数指定的并行度数量
parallelism.default: 1

# 重启策略
jobmanager.execution.failover-strategy: region

# 存储临时文件的路径，如果没有配置，则默认采用服务器的临时目录，如 LInux 的 /tmp 目录
io.tmp.dirs: /tmp

参考Flink的官方手册：更多配置

配置历史服务器

概述

运行Flink job的集群一旦停止，只能去yarn或本地磁盘上查看日志，对于Job任务信息的查看、异常问题的排查非常不友好。

Flink提供了历史服务器，用来在相应的Flink集群关闭后查询已完成作业的统计信息。通过History Server可以查询这些已完成作业的统计信息，无论是正常退出还是异常退出。

Flink任务停止后，JobManager会将已经完成任务的统计信息进行存档，History Server进程则在任务停止后可以对任务统计信息进行查询。

配置

创建存储目录

[root@node01 flink]# hadoop fs -mkdir -p /logs/flink-job

在flink-config.yaml中添加如下配置

#==============================================================================
# HistoryServer
#==============================================================================

# The HistoryServer is started and stopped via bin/historyserver.sh (start|stop)

# Directory to upload completed jobs to. Add this directory to the list of
# monitored directories of the HistoryServer as well (see below).
#jobmanager.archive.fs.dir: hdfs:///completed-jobs/
jobmanager.archive.fs.dir: hdfs://node01:9000/logs/flink-job

# The address under which the web-based HistoryServer listens.
#historyserver.web.address: 0.0.0.0
historyserver.web.address: node01

# The port under which the web-based HistoryServer listens.
#historyserver.web.port: 8082
historyserver.web.port: 8082

# Comma separated list of directories to monitor for completed jobs.
#historyserver.archive.fs.dir: hdfs:///completed-jobs/
historyserver.archive.fs.dir: hdfs://node01:9000/logs/flink-job

# Interval in milliseconds for refreshing the monitored directories.
#historyserver.archive.fs.refresh-interval: 10000
historyserver.archive.fs.refresh-interval: 5000

启动、停止历史服务器

启动历史服务器

[root@node01 flink]# bin/historyserver.sh start
Starting historyserver daemon on host node01.

停止历史服务器

[root@node01 flink]# bin/historyserver.sh stop
Stopping historyserver daemon (pid: 30749) on host node01.

提交一个Job任务

[root@node01 flink]# bin/flink run -t yarn-per-job -c com.atguigu.wc.WordCountStreamUnboundedDemo  /root/FlinkTutorial-1.17-1.0-SNAPSHOT.jar

2023-06-12 23:41:00,719 INFO  org.apache.hadoop.hdfs.protocol.datatransfer.sasl.SaslDataTransferClient [] - SASL encryption trust check: localHostTrusted = false, remoteHostTrusted = false
2023-06-12 23:41:00,742 INFO  org.apache.hadoop.hdfs.protocol.datatransfer.sasl.SaslDataTransferClient [] - SASL encryption trust check: localHostTrusted = false, remoteHostTrusted = false
2023-06-12 23:41:00,761 INFO  org.apache.flink.yarn.YarnClusterDescriptor                  [] - Cannot use kerberos delegation token manager, no valid kerberos credentials provided.
2023-06-12 23:41:00,766 INFO  org.apache.flink.yarn.YarnClusterDescriptor                  [] - Submitting application master application_1686577483648_0012
2023-06-12 23:41:00,792 INFO  org.apache.hadoop.yarn.client.api.impl.YarnClientImpl        [] - Submitted application application_1686577483648_0012
2023-06-12 23:41:00,792 INFO  org.apache.flink.yarn.YarnClusterDescriptor                  [] - Waiting for the cluster to be allocated
2023-06-12 23:41:00,793 INFO  org.apache.flink.yarn.YarnClusterDescriptor                  [] - Deploying cluster, current state ACCEPTED
2023-06-12 23:41:04,565 INFO  org.apache.flink.yarn.YarnClusterDescriptor                  [] - YARN application has been deployed successfully.
2023-06-12 23:41:04,565 INFO  org.apache.flink.yarn.YarnClusterDescriptor                  [] - Found Web Interface node02:38887 of application 'application_1686577483648_0012'.
Job has been submitted with JobID cd41d983c93d8eb906c9aa899dcdefd0

访问http://node01:8088/cluster查看Hadoop 在这里插入图片描述 访问Web UI查看提交任务信息 在这里插入图片描述

查看历史Job信息

在浏览器地址栏输入：http://node01:8082 查看已经停止的 job 的统计信息 在这里插入图片描述 停止提交任务

[root@node01 flink]# bin/flink cancel -t yarn-per-job -Dyarn.application.id=application_1686577483648_0012 cd41d983c93d8eb906c9aa899dcdefd0

访问http://node01:9870/explorer.html#/logs/flink-job查看HDFS中的归档文件 在这里插入图片描述 等一段时间，几分钟后查看历史服务器 在这里插入图片描述 查看Job具体信息 在这里插入图片描述

标签：node02,flink,node01,Flink,host,集群,HA,root
From： https://blog.51cto.com/chencoding/7872587

相关文章

  • ES集群故障转移
    6ES集群故障转移6.1什么是故障转移所谓故障转移指的是，当集群中有节点发生故障时，这个集群是如何进行自动修复的。ES集群目前是由3个节点组成，如下图所示，此时集群状态是green6.2模拟节点故障假设：node1所在机器宕机导致服务终止，此时集群会如何处理；大体分为三个步骤：1.重新选举2.......
  • The 2021 CCPC Weihai Onsite
    Preface又被打爆了，看了下榜这场罚时比较炸喜提银首咯不过yysy这场题出的还是挺好的，medium题都挺有意思需要想一想但就是感觉考的组合计数这一块有点太多了，而且因为有人歪榜开局过了M，导致我前期一直在这道题上坐牢，最后还是徐神出马一套生成函数秒了此题A.Goodbye,Ziyin!签......
  • hashmap,arrayList,concurrentHashMap扩容机制
    HashMap1.7和1.8扩容机制在Java1.7中，HashMap的扩容机制是当容量超过负载因子与数组长度的乘积时就会进行扩容。默认负载因子为0.75，即当数组长度为n时，当元素个数size超过n*0.75时就会扩容。扩容时，数组长度会变为原来的2倍，并且将原来的元素重新计算哈希值，再散列到新......
  • Dual Graph enhanced Embedding Neural Network for CTR Prediction
    目录概DG-ENNGuoW.,SuR.,TanR.,GuoH.,ZhangY.,LiuZ.,TangR.andHeX.Dualgraphenhancedembeddingneuralnetworkforctrprediction.KDD,2021.概图网络用在精排上,作者的出发点是为了解决(user/item)特征的稀疏性和用户交互序列的稀疏性,不过这出......
  • 记Halo1.5版本迁移Halo2.10.0版本
    原文地址:记Halo1.5版本迁移Halo2.10.0版本-Stars-One的杂货小窝上一篇Window10安装linux子系统及子系统安装1Panel面板-Stars-One的杂货小窝将对应的环境都准备好了,这下我们可以通过1Panel进行halo的2.10版本安装,来测试新版本升级是否有什么问题本篇文章实际是在本地L......
  • Adobe Character Animator（CH版）2020 Mac 软件下载 永久使用
    AdobeCharacterAnimator（CH版）2020Mac 是一款动作捕获和角色动画制作工具，允许您使用自己的图稿创造富有表现力的角色，实时将2D人物制成动画，创造人物并用您自己的表演让它动起来。CharacterAnimator使用您的实时表情和动作将人物动起来。因此，当您微笑或点头时，您的人物也会做出......
  • MySQL错误：check the manual that corresponds to your MySQL server version for the
    在MySQL执行以下SQL报错DELIMITER//CREATEPROCEDUREgenerate_and_insert_data()BEGINDECLAREiINTDEFAULT1;DECLAREjINTDEFAULT1;DECLAREtotal_iterationsINTDEFAULT1000;WHILEi<=total_iterationsDO--创建临时表用于存储生成......
  • ASP.NET Core使用Hangfire定时发布文章
    ASP.NETCore使用Hangfire实现定时任务前言也是上了5天班，终于迎来了休息，抽空更新下博客，然后就是下周一公司会对我进行考核，希望考核能通过吧！！！然后我想给博客添加一个定时发布文章的功能，其实这个功能对于我的博客是没什么作用的，什么时候发都没什么人看。但是咱还是要有这个功能。......
  • 使用Pycharm环境，搭建Django开发环境
    早期都会直接在Linux系统下，直接把很多py包装到物理系统里。不过事实证明，装在虚拟环境里要方便得多，尤其是早期很多工程做到一半发现写不下去了。假如IDE选择的是Pycharm的话，实际上在它的工具里，就非常方便了。 新建一个工程后，在Pycharm里，点击如上图所示的蓝色按钮，就进入了虚拟......
  • How can I change the reference numbers in manuscript to blue color?
    HowcanIchangethereferencenumbersinmanuscripttobluecolor?IamworkinginWord2010and EndNoteX7.Iwanttochangethecolorofcitationsinmanuscripttoblue(nottochangethemmanually), buttochangethemautomaticallyincreatingt......

赞助商

阅读排行

网站主页关于我们联系我们网站地图

本网站内容转载自其他媒体，侵权联系[admin##ips99.com]。

Copyright © 2020-2023 IPS99 版权所有 IPS99