flink 集群部署_flink集群部署

1. 环境准备

准备三台主机：可以参考：linux-centos7 集群环境搭建

mitchell-101

mitchell-102

mitchell-103

2. 下载安装包

这里以 1.14.6 版本为例：点击下载

其他历史版本的下载方式，可以打开下方地址：

https://www.apache.org/dyn/closer.lua/flink/flink-{version}/flink-{version}-bin-scala_2.12.tgz

其中 {version} 替换为具体的版本号：如需要下载 1.14.6

https://www.apache.org/dyn/closer.lua/flink/flink-1.14.6/flink-1.14.6-bin-scala_2.12.tgz

3. 安装

上传安装包到服务器的 /opt/software 目录

、

解压安装包，并放置到安装目录 /opt/module

cd /opt/software
tar -zxvf flink-1.14.6-bin-scala_2.12.tgz -C /opt/module/
cd /opt/module
mv flink-1.14.6 flink
1
2
3
4

3.1. Standalone 集群部署

Session 模式

3.1.1. 修改配置文件

cd flink/conf
1

vim flink-conf.yaml

修改下方通用的配置内容

# jobmanager 外部访问的地址 和 端口
jobmanager.rpc.address: mitchell-101
jobmanager.rpc.port: 6123

# jobmanager 占用的总内存大小，包含 JVM，不能太小，否则集群无法启动
jobmanager.memory.process.size: 1600m

# taskmanager 占用的总内存大小，包含 JVM，不能太小，否则集群无法启动
taskmanager.memory.process.size: 1600m

# taskmanager 每个实例的插槽数量
taskmanager.numberOfTaskSlots: 1

# 未设置并行度时的默认并行度
parallelism.default: 1
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15

vim masters

配置 jobmanager

mitchell-101:8081
1

vim workers

配置 taskmanager

mitchell-101
mitchell-102
mitchell-103
1
2
3

3.1.2. 启动、关闭、访问

分发项目到其它服务器

cd /opt/module
xsync.sh flink
cd /opt/module/flink
1
2
3

启动和关闭集群

./bin/start-cluster.sh
./bin/stop-cluster.sh
1
2

启动后访问地址：http://mitchell-101:8081 出现下方界面则为部署成功

为了操作方便编写了关于启动和关闭的 shell 脚本：mflink.sh

#!/bin/bash

if [ $# -lt 1 ]
then
    echo "请输入参数 start 或者 stop"
    exit;
fi

case $1 in
"start")
    echo "=============================== 启动 flink 集群 ==============================="
    /opt/module/flink/bin/start-cluster.sh
;;
"stop")
    echo "=============================== 关闭 flink 集群 ==============================="
    /opt/module/flink/bin/stop-cluster.sh
;;
*)
    echo "请输入参数 start 或 stop"
;;
esac
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21

3.2. Yarn 集群部署（生产推荐）

该集群方式首先需要部署 Hadoop，部署方式参考：hadoop3 完全分布式安装

该模式不需要每台机器都有 Flink 包，只需要你执行命令的那台机器有程序包即可

并且执行命令的机器必须是 NameNode 所在的服务器

PS：注意下方的模式不适用于 HA 高可用模式的 Hadoop 集群

3.2.1. 配置环境变量

# HADDOP
export HADOOP_HOME=/opt/module/hadoop-3.3.4
export PATH=$PATH:$HADOOP_HOME/bin:$HADOOP_HOME/sbin
export HADOOP_CONF_DIR=$HADOOP_HOME/etc/hadoop
export HADOOP_CLASSPATH=`hadoop classpath`
1
2
3
4
5

3.2.2. 启动 Session 模式

该模式使用一个已经运行的集群来执行所有提交的应用程序。该集群中执行的应用程序竞争使用该集群的资源。这样做的好处是，不必为每个提交的作业启动一个集群，额外增加资源开销。但是，该模式下，如果集群中的一个作业行为不正常或导致 TaskManager 宕机，则在该 TaskManager 上运行的所有作业都将受到故障的影响。也就是说，一个作业也会影响其他作业可能导致大规模的重启。所有重新启动的作业都会同时访问文件系统，可能导致其他服务无法使用该文件系统。此外，让单个集群运行多个作业意味着 JobManager 的负载更大。该模式适合于对启动延迟要求较高且运行时间较短的作业，例如交互式查询。

• ./bin/yarn-session.sh：启动文件
• -d：后台运行
• -jm 1024：jobmanger 占用内存大小为 1024 M
• -tm 1024：taskmanger 占用的内存大小 为 1024M
• -s 2：每个 taskmanger 的槽位大小 为 2
• -m yarn-cluster：指定为 yarn 集群模式
• -nm test：指定任务名称为 test
• -qu default：指定使用 yarn 默认的队列模式

# 启动集群
./bin/yarn-session.sh -d -jm 1024 -tm 1024 -s 2 -m yarn-cluster -nm test -qu default

# 关闭集群
yarn application -kill [启动时 YARN 分配的集群唯一标识]
举例：yarn application -kill application_1699810464399_0003

# 提交任务，需要先启动集群
./bin/flink run -c [主类全路径] [jar包路径]
举例：./bin flink -c com.mitchell.mus.flink.F01WordCount ./lib/mus-flink.jar
1
2
3
4
5
6
7
8
9
10

3.2.3. 启动 Per-Job 模式

该模式下，集群管理框架 YARN 会为每个提交的作业启动一个单独Flink集群，该集群仅对该作业可用。作业运行完成，集群将被关闭，所有用到的资源（例如文件）也将被清理。此模式提供更好的资源隔离。一个作业故障不会影响其他作业的运行。此外，由于每个作业都有自己的 JobManager，负载被分散到多个实体上。考虑到 Session 模式的资源隔离问题，对于需要长时间运行的作业，通常选择 Per-Job 模式，这些作业愿意承受增加启动延迟以提升作业的恢复能力。

总之，在 Session 模式下，集群生命周期独立于集群上运行的任何作业，并且集群上运行的所有作业共享集群资源。Per-Job 模式为每个提交的作业启动一个集群，以提供更好的资源隔离保证。在这种情况下，集群的生命周期与作业的生命周期绑定。

• ./bin/flink：启动文件
• run：运行
• -d：后台运行
• -m yarn-cluster：指定为 yarn 集群模式
• -yjm 1024：jobmanger 占用内存大小为 1024 M
• -ytm 1024：taskmanger 占用的内存大小 为 1024M
• -ynm test：指定任务名称为 test
• -c com.mitchell.mus.flink.F01WordCount ./lib/mus-flink.jar：指定 Main 方法全路径类 以及 jar 包

./bin/flink run -d -m yarn-cluster -yjm 1024 -ytm 1024 -ynm test -c com.mitchell.mus.flink.F01WordCount ./lib/mus-flink.jar
1

3.2.4. 启动 Application 模式

Application 模式将 Per-Job 模式的资源隔离与轻量级、可扩展的作业提交过程结合起来。Application 模式为每个提交的作业创建一个集群，但作业的main 方法将在 JobManager 上执行，以节省提取 job graph 所需的 CPU 周期，也节省客户端下载依赖及将 job graph 及其依赖关系传送到群集所需的带宽。

• ./bin/flink：启动文件
• run-application：应用模式运行
• -t yarn-application：yarn 引用模式
• -c com.mitchell.mus.flink.F01WordCount ./lib/mus-flink.jar：指定 Main 方法全路径类 以及 jar 包

./bin/flink run-application -t yarn-application -c com.mitchell.mus.flink.F01WordCount ./lib/mus-flink.jar
1

我们也可以通过-D传入一些配置项：如 设置 内存大小

具体可以配置哪些参数可以参考配置文件：flink-1.14.6/conf/flink-conf.yaml

-Djobmanager.memory.process.size=1024m -Dtaskmanager.memory.process.size=1024m
1

3.2.5. 界面效果