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

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

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

2000元阿里云代金券免费领取，2核4G云服务器仅664元/3年，新老用户都有优惠，立即抢购>>>


阿里云采购季（云主机223元/3年）活动入口：请点击进入>>>，


阿里云学生服务器（9.5元/月）购买入口：请点击进入>>>，

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
资源调度
存储
关键词：
实时计算 Flink版配置
云服务器 ECS flink
flink云服务器 ECS
实时计算 Flink版服务器
实时计算 Flink版集群
相关实践学习
基于Hologres轻松玩转一站式实时仓库
本场景介绍如何利用阿里云MaxCompute、实时计算Flink和交互式分析服务Hologres开发离线、实时数据融合分析的数据大屏应用。
Linux入门到精通
本套课程是从入门开始的Linux学习课程，适合初学者阅读。由浅入深案例丰富，通俗易懂。主要涉及基础的系统操作以及工作中常用的各种服务软件的应用、部署和优化。即使是零基础的学员，只要能够坚持把所有章节都学完，也一定会受益匪浅。
CodeDevMaster
目录
相关文章
爱吃糖的范同学
|
5天前
|
资源调度 监控 数据处理
【Flink】Flink集群有哪些角色？各自有什么作用？
【4月更文挑战第18天】【Flink】Flink集群有哪些角色？各自有什么作用？
爱吃糖的范同学
79 7
cuicuicuic
|
3天前
|
SQL 分布式计算 关系型数据库
实时计算 Flink版产品使用合集之MySQL CDC Connector是否需要在Flink服务器上单独部署
实时计算Flink版作为一种强大的流处理和批处理统一的计算框架，广泛应用于各种需要实时数据处理和分析的场景。实时计算Flink版通常结合SQL接口、DataStream API、以及与上下游数据源和存储系统的丰富连接器，提供了一套全面的解决方案，以应对各种实时计算需求。其低延迟、高吞吐、容错性强的特点，使其成为众多企业和组织实时数据处理首选的技术平台。以下是实时计算Flink版的一些典型使用合集。
cuicuicuic
11 0
cuicuicuic
|
3天前
|
SQL 弹性计算 分布式计算
实时计算 Flink版产品使用合集之如果产品是基于ak的，可以提交sql任务到ecs自建hadoop集群吗
实时计算Flink版作为一种强大的流处理和批处理统一的计算框架，广泛应用于各种需要实时数据处理和分析的场景。实时计算Flink版通常结合SQL接口、DataStream API、以及与上下游数据源和存储系统的丰富连接器，提供了一套全面的解决方案，以应对各种实时计算需求。其低延迟、高吞吐、容错性强的特点，使其成为众多企业和组织实时数据处理首选的技术平台。以下是实时计算Flink版的一些典型使用合集。
cuicuicuic
15 0
cuicuicuic
|
3天前
|
资源调度 Kubernetes Oracle
实时计算 Flink版产品使用合集之三种集群模式各有啥优缺点，生产环境如何选择
实时计算Flink版作为一种强大的流处理和批处理统一的计算框架，广泛应用于各种需要实时数据处理和分析的场景。实时计算Flink版通常结合SQL接口、DataStream API、以及与上下游数据源和存储系统的丰富连接器，提供了一套全面的解决方案，以应对各种实时计算需求。其低延迟、高吞吐、容错性强的特点，使其成为众多企业和组织实时数据处理首选的技术平台。以下是实时计算Flink版的一些典型使用合集。
cuicuicuic
19 0
cuicuicuic
|
3天前
|
SQL Java 数据处理
实时计算 Flink版产品使用合集之Session集群启动后一直不可用时如何解决
实时计算Flink版作为一种强大的流处理和批处理统一的计算框架，广泛应用于各种需要实时数据处理和分析的场景。实时计算Flink版通常结合SQL接口、DataStream API、以及与上下游数据源和存储系统的丰富连接器，提供了一套全面的解决方案，以应对各种实时计算需求。其低延迟、高吞吐、容错性强的特点，使其成为众多企业和组织实时数据处理首选的技术平台。以下是实时计算Flink版的一些典型使用合集。
cuicuicuic
12 1
三分钟热度的鱼
|
5天前
|
机器学习/深度学习 人工智能 流计算
人工智能平台PAI 操作报错合集之在集群上提交了包含alink相关功能的flink任务，但是却报错如何解决
阿里云人工智能平台PAI (Platform for Artificial Intelligence) 是阿里云推出的一套全面、易用的机器学习和深度学习平台，旨在帮助企业、开发者和数据科学家快速构建、训练、部署和管理人工智能模型。在使用阿里云人工智能平台PAI进行操作时，可能会遇到各种类型的错误。以下列举了一些常见的报错情况及其可能的原因和解决方法。
三分钟热度的鱼
45 1
yuanzhengme
|
5天前
|
存储 Java 网络安全
ZooKeeper【搭建 03】apache-zookeeper-3.6.0 伪集群版（一台服务器实现三个节点的ZooKeeper集群）
【4月更文挑战第10天】ZooKeeper【搭建 03】apache-zookeeper-3.6.0 伪集群版（一台服务器实现三个节点的ZooKeeper集群）
yuanzhengme
35 1
一只牛博
|
5天前
|
Java 网络安全 Apache
搭建Zookeeper集群：三台服务器，一场分布式之舞
搭建Zookeeper集群：三台服务器，一场分布式之舞
一只牛博
53 0
xin在这
|
1天前
|
Oracle 关系型数据库 数据库
实时计算 Flink版操作报错合集之执行Flink job，报错“Could not execute SQL statement. Reason:org.apache.flink.table.api.ValidationException: One or more required options are missing”，该怎么办
在使用实时计算Flink版过程中，可能会遇到各种错误，了解这些错误的原因及解决方法对于高效排错至关重要。针对具体问题，查看Flink的日志是关键，它们通常会提供更详细的错误信息和堆栈跟踪，有助于定位问题。此外，Flink社区文档和官方论坛也是寻求帮助的好去处。以下是一些常见的操作报错及其可能的原因与解决策略。
xin在这
8 0
xin在这
|
2天前
|
消息中间件 关系型数据库 MySQL
实时计算 Flink版操作报错合集之遇到报错:Apache Kafka Connect错误如何解决
在使用实时计算Flink版过程中，可能会遇到各种错误，了解这些错误的原因及解决方法对于高效排错至关重要。针对具体问题，查看Flink的日志是关键，它们通常会提供更详细的错误信息和堆栈跟踪，有助于定位问题。此外，Flink社区文档和官方论坛也是寻求帮助的好去处。以下是一些常见的操作报错及其可能的原因与解决策略。
xin在这
13 5

热门文章

最新文章

  • 1
    阿里云学生云服务器申请，阿里云送每个大学生一台云服务器
  • 2
    Serverless 应用引擎操作报错合集之在阿里云函数计算中，服务器调用FC函数时出现 "[Errno -3] Temporary failure in name resolution)" 错误如何解决
  • 3
    如何在阿里云服务器快速搭建部署Nginx环境
  • 4
    阿里云服务器搭建部署宝塔详细流程
  • 5
    阿里云服务器购买之后发票如何申请？申请发票流程及常见问题参考
  • 6
    Apsara Clouder云计算专项技能认证:云服务器ECS
  • 7
    阿里云服务器搭建部署宝塔详细流程
  • 8
    Linux系统之安装ServerBee服务器监控工具
  • 9
    在阿里云服务器上部署Tomcat详细图文详解
  • 10
    阿里云服务器经济型e、通用算力型u1与c7/g7/r7/c8y/g8y/r8y实例区别及选择参考
  • 1
    Flink CDC产品常见问题之look up hint 没有生效如何解决
    820
  • 2
    Flink CDC产品常见问题之tidb cdc 数据量大了就疯狂报空指针如何解决
    451
  • 3
    Apache Flink官方网站提供了关于如何使用Docker进行Flink CDC测试的文档
    349
  • 4
    多个2.4的Flink CDC会有debezium冲突怎么解决啊？
    29
  • 5
    Flink CDC产品常见问题之使用3.0测试mysql到starrocks启动报错如何解决
    119
  • 6
    flink cdc 转换问题之类型转换如何解决
    67
  • 7
    Flink CDC产品常见问题之oracel cdc 的延时很大如何解决
    38
  • 8
    flink cdc 数据问题之数据丢失如何解决
    155
  • 9
    Flink CDC产品常见问题之把flink cdc同步的数据写入到目标服务器失败如何解决
    37
  • 10
    Flink CDC产品常见问题之tidb-HUDI的整库同步失败如何解决
    49

    • 相关课程

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

    • 相关电子书

    更多
  • Flink CDC Meetup PPT - 龚中强
  • Flink CDC Meetup PPT - 王赫
  • Flink CDC Meetup PPT - 覃立辉

    • 相关实验场景

    更多
  • 如何开通MongoDB的专属集群
  • ACK集群应用部署进阶
  • 使用阿里云Elasticsearch快速搭建可观测系统
  • 体验PolarDB MySQL多主集群
  • 部署高可用架构
    • 下一篇
    2024年阿里云免费云服务器及学生云服务器申请教程参考
    http://www.vxiaotou.com