AI 助理
备案控制台
开发者社区 大数据 文章 正文

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

2023-09-21 1098
版权
版权声明：
本文内容由阿里云实名注册用户自发贡献，版权归原作者所有，阿里云开发者社区不拥有其著作权，亦不承担相应法律责任。具体规则请查看《 阿里云开发者社区用户服务协议》和 《阿里云开发者社区知识产权保护指引》。如果您发现本社区中有涉嫌抄袭的内容，填写 侵权投诉表单进行举报，一经查实，本社区将立刻删除涉嫌侵权内容。
简介： 本文介绍了如何搭建一个Flink集群、Flink集群HA高可用，并配置历史服务器以记录Job任务执行的详细信息和状态。

Flink集群搭建

集群规划

节点 node01 node02 node03
角色 JobManager
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集群和任务进行监控管理。

image.png

注意：关闭防火墙，否则可能无法访问，或者集群的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
TaskManager		 JobManager
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
image.png

访问http://node02:8081
image.png

测试

查看ZK：JobManager节点信息
image.png

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是否变化
image.png

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

image.png

访问Web UI查看提交任务信息
image.png

查看历史Job信息

在浏览器地址栏输入：http://node01:8082 查看已经停止的 job 的统计信息
image.png

停止提交任务

[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中的归档文件
image.png

等一段时间，几分钟后查看历史服务器
image.png

查看Job具体信息
image.png

文章标签：
实时计算 Flink版
微服务引擎
流计算
分布式计算
Hadoop
资源调度
存储
关键词：
云服务器 ECS配置
配置云服务器 ECS
云服务器 ECS集群
高可用云服务器 ECS
实时计算 Flink版集群
相关实践学习
基于Hologres+Flink搭建GitHub实时数据大屏
通过使用Flink、Hologres构建实时数仓，并通过Hologres对接BI分析工具（以DataV为例），实现海量数据实时分析.
实时计算 Flink 实战课程
如何使用实时计算 Flink 搞定数据处理难题？实时计算 Flink 极客训练营产品、技术专家齐上阵，从开源 Flink功能介绍到实时计算 Flink 优势详解，现场实操，5天即可上手！ 欢迎开通实时计算 Flink 版： https://cn.aliyun.com/product/bigdata/sc Flink Forward Asia 介绍： Flink Forward 是由 Apache 官方授权，Apache Flink Community China 支持的会议，通过参会不仅可以了解到 Flink 社区的最新动态和发展计划，还可以了解到国内外一线大厂围绕 Flink 生态的生产实践经验，是 Flink 开发者和使用者不可错过的盛会。 去年经过品牌升级后的 Flink Forward Asia 吸引了超过2000人线下参与，一举成为国内最大的 Apache 顶级项目会议。结合2020年的特殊情况，Flink Forward Asia 2020 将在12月26日以线上峰会的形式与大家见面。
CodeDevMaster
目录
相关文章
弹性计算小冉
|
3月前
|
存储 人工智能 容灾
阿里云服务器2核8G、4核16G、8核32G配置热门实例性能对比与场景化选型指南
2核8G/4核16G/8核32G配置的阿里云服务器在阿里云活动中目前有经济型e、通用算力型u1、通用型g7、通用型g8y和通用型g9i五种实例可选，目前2核8G配置选择u1实例活动价格652.32元1年起，4核16G月付选择经济型e实例最低89元1个月，8核32G配置160元1个月起，本文将为大家解析经济型e、通用算力型u1、通用型g7及通用型g8y实例，帮助用户根据自身需求合理选择最适合的实例规格和配置。
弹性计算小冉
630 6
上云百科
|
2月前
|
弹性计算 ice
阿里云4核8G云服务器配置价格：热门ECS实例及CPU处理器型号说明
阿里云2025年4核8G服务器配置价格汇总，涵盖经济型e实例、计算型c9i等热门ECS实例，CPU含Intel Xeon及AMD EPYC系列，月费159元起，年付低至1578元，按小时计费0.45元起，实际购买享折扣优惠。
上云百科
533 1
弹性计算小冉
|
3月前
|
存储 运维 安全
阿里云服务器2核4G与4核8G配置解析：实例规格、价格及选择指南
阿里云服务器2核4G和4核8G配置凭借其均衡的性能与性价比，成为中小企业及开发者群体的热门选择。通用算力型u1实例2核4G5M带宽80G ESSD Entry云盘特惠价199元1年，个人用户最低531.79元1年，4核8G配置价格最低的是通用算力型u1实例，活动价格为955.58元1年起，除此之外，还有计算型c7、c9i、c8y等计算型实例有2核4G与4核8G配置可选。具体配置还需要看想要购买的云服务器实例规格和配置及带宽大小。本文将结合2025年最新活动价格，深度解析阿里云服务器2核4G与4核8G配置的实例规格、适用场景及选择策略，以供参考。
弹性计算小冉
368 1
弹性计算小冉
|
3月前
|
机器学习/深度学习 存储 前端开发
阿里云服务器8核16G配置详解：热门实例规格、价格与选购指南
阿里云服务器8核16G配置，这样的配置能够满足大多数中小型网站、Web前端服务器以及计算密集型任务的需求。根据最新的收费标准及活动价格来看，云服务器8核16G配置收费标准最低的是经济型e实例，按量收费标准0.9元/小时，如果是需要年付的话，目前选择计算型c8y实例的活动价格为为2861.27元/1年起。本文为大家介绍8核16G配置的几款主要热门实例规格各自的性能与收费标准价格情况，以供选购参考。
弹性计算小冉
556 3
程序员在线
|
2月前
|
弹性计算 定位技术 数据中心
阿里云服务器配置选择方法：付费类型、地域及CPU内存配置全解析
阿里云服务器怎么选？2025最新指南：就近选择地域，降低延迟；长期使用选包年包月，短期灵活选按量付费；企业选2核4G5M仅199元/年，个人选2核2G3M低至99元/年，高性价比爆款推荐，轻松上云。
程序员在线
188 11
云服务器小助手
|
2月前
|
存储 弹性计算 网络协议
超详细的阿里云服务器购买流程，ECS自定义购买配置教程
本文详细图解阿里云ECS服务器自定义购买全流程，涵盖付费模式、地域选择、网络配置、实例规格、镜像、存储、安全组及登录设置等核心步骤，助您轻松掌握专业级云服务器搭建方法。
云服务器小助手
326 10
翼龙云TG_yilongcloud
|
2月前
|
存储 弹性计算 监控
阿里云渠道商：如何挑选阿里云服务器配置？
本文详解通用型、计算型、内存型等实例适用场景，结合性能评估与成本优化策略，助力用户按需选择。以日均1万访问企业网站为例，2核4G+3M带宽月费约200元，性价比高。合理配置更省钱。
翼龙云TG_yilongcloud
128 1
游客s65ycxi6wnv5q
|
2月前
|
弹性计算
阿里云ECS云服务器8核16G配置收费价格，多种ECS实例CPU及费用清单
阿里云8核16G云服务器价格因实例类型而异。计算型c9i约743元/月，一年6450元（7折）；通用算力型u1仅673元/月，一年4225元（5.1折）。实际价格享时长折扣，详情见ECS官网。
游客s65ycxi6wnv5q
419 3

热门文章

最新文章

  • 1
    【湖仓一体】阿里云AnalyticDB MySQL基于Flink CDC+Hudi实现多表全增量入湖实践
  • 2
    Flink 靠什么征服饿了么工程师？
  • 3
    史上超强阵容！大数据及人工智能领域顶级盛会，Flink Forward Asia 2019 不容错过！
  • 4
    Apache Flink 漫谈系列(08) - SQL概览
  • 5
    快手基于 Flink 的持续优化与实践
  • 6
    Flink 实时计算在微博的应用
  • 7
    Airwallex 基于 Flink 打造实时风控系统
  • 8
    Flink 容错恢复 2.0 2022 最新进展
  • 9
    《Apache Flink 案例集（2022版）》——2.数据分析——美团-Flink 的实时数仓平台建设（2）
  • 10
    Flink常用Sink(elasticsearch(es)Sink、RedisSink、KafkaSink、MysqlSink、FileSink）
  • 1
    阿里云ECS的使用心得
    134
  • 2
    阿里云ECS使用体验
    179
  • 3
    pycharm连接云服务器#潞晨cloud
    370
  • 4
    构建高效稳定的Linux服务器：系统优化与安全策略
    255
  • 5
    ECS使用体验
    115
  • 6
    阿里云ECS的使用心得
    69
  • 7
    蓝易云 - 比较分析香港服务器有哪些优势
    324
  • 8
    服务器数据恢复—误删除VMware虚拟机vmdk文件的数据恢复案例
    481
  • 9
    阿里云服务器怎么开发票？图文教程来了
    1726
  • 10
    【阿里云弹性计算】ECS实例的生命周期管理：阿里云自动化工具与策略介绍
    295

    • 相关课程

    更多
  • 大数据Flink实时旅游平台环境篇 2020版
  • Apache Flink 入门到实战 - Flink开源社区出品
  • 实时计算 Flink 版产品入门与实操
  • 开源 Flink 极速上手教程
  • 大数据实时计算框架Spark快速入门
  • Apache Flink 入门

    • 相关电子书

    更多
  • Flink峰会 - 李佳林
  • Flink峰会 - 徐榜江
  • Flink CDC Meetup PPT - 龚中强

    • 相关实验场景

    更多
  • 基于Hologres+Flink搭建GitHub实时数据大屏
  • 基于Flink+Kafka实现订单评论实时分析
    • 下一篇
    「直播预告」Streaming Lakehouse Meetup EP.2｜Paimon × StarRocks 共话实时湖仓