Linux软件安装 —— Flink集群安装（集成Zookeeper、Hadoop高可用）

文章目录

• 一、节点说明
• 二、配置节点间免密登录
• 三、JDK安装
• 四、Zookeeper安装
• 五、Hadoop安装
• 六、Flink安装
• • • 1、基础环境准备
      • • • （1）下载安装包
            * （2）上传并解压
      • 2、修改配置
      • • • （1）配置zookeeper
            * （2）配置flink-conf.yaml
            * （3）配置workers
            * （4）创建必要的目录
            * （5）配置环境变量
      • 3、分发flink
• 七、集群测试
• • • 1、启动zookeeper，hadoop
      • 2、Yarn Session测试
      • • • （1）模式介绍
            * （2）准备测试资源
            * （3）启动YARN Session
            * （4）查看YARN Session状态
            * （5）提交作业
            * （6）停止YARN Session
      • 3、Per-Job测试
      • • • （1）Per-Job模式介绍
            * （2）提交per-job作业

一、节点说明

1、相关软件

2、相关进程

二、配置节点间免密登录

参考本人此篇文章：Linux软件安装 —— SSH免密登录

三、JDK安装

参考本人此篇文章：Linux软件安装 —— JDK安装

四、Zookeeper安装

参考本人此篇文章：Linux软件安装 —— zookeeper集群安装

五、Hadoop安装

参考本人此篇文章：Linux软件安装 —— Hadoop高可用安装（集成Zookeeper）

六、Flink安装

1、基础环境准备

（1）下载安装包

官方文档：Downloads | Apache Flink 一直拉倒最后就会显示所有历史版本

本文使用版本为：flink-1.13.6-bin-scala_2.12.tgz

（2）上传并解压

1# 加压到安装目录
2tar -zxvf flink-1.13.6-bin-scala_2.12.tgz -C /opt/module/
3
4# 修改名称
5mv flink-1.13.6/ flink
6

2、修改配置

（1）配置zookeeper

使用flink内置zookeeper才配，自己搭的不用配此项

1# 编辑flink中的zookeeper文件
2vim /opt/module/flink/conf/zoo.cfg
3
4# 添加data和log位置
5dataDir=/opt/module/flink/flink-zookeeper/data
6dataLogDir=/opt/module/flink/flink-zookeeper/logs
7
8# 修改zookeeper集群信息
9server.2=node02:2888:3888
10server.3=node03:2888:3888
11server.4=node04:2888:3888
12

（2）配置flink-conf.yaml

1# 备份原始文件
2cp flink-conf.yaml flink-conf.yaml.bak
3
4# 编辑flink-conf.yaml文件
5vim /opt/module/flink/conf/flink-conf.yaml
6
7# ==================== 基础配置 ====================
8# JobManager节点，指定node02为JobManager
9jobmanager.rpc.address: node02
10jobmanager.rpc.port: 6123
11# JobManager堆内存（根据实际内存调整，建议4G以上）
12jobmanager.memory.process.size: 1024m
13# TaskManager堆内存（根据实际内存调整）
14taskmanager.memory.process.size: 1024m
15# 每个TaskManager的slot数量（根据CPU核心数调整）
16taskmanager.numberOfTaskSlots: 1
17# 并行度默认值
18parallelism.default: 1
19web.tmpdir: /opt/module/flink/flink-jar
20blob.storage.directory: /opt/module/flink/flink-blob
21yarn.maximum-failed-containers: 200
22taskmanager.tmp.dirs: /opt/module/flink/flink-data
23flink_log_bak_dir: /opt/module/flink/logs
24flink_log_dir: /opt/module/flink/logs
25
26# ==================== 高可用配置 ====================
27high-availability: zookeeper
28high-availability.zookeeper.quorum: node02:2181,node03:2181,node04:2181
29# Zookeeper中Flink的根路径
30high-availability.zookeeper.path.root: /flink
31# JobManager元数据存储路径（使用HDFS）
32high-availability.zookeeper.storageDir: hdfs:///flink/recovery
33fs.hdfs.hadoopconf: /opt/module/hadoop/etc/hadoop
34fs.hdfs.hdfssite: /opt/module/hadoop/etc/hadoop/hdfs-site.xml
35
36# ==================== 故障恢复、checkpoint ====================
37restart-strategy: fixed-delay
38restart-strategy.fixed-delay.attempts: 3
39restart-strategy.fixed-delay.delay: 5 s
40state.backend: rocksdb
41state.backend.incremental: true
42state.checkpoint-storage: filesystem
43state.checkpoints.dir: hdfs:///flink/flink-checkpoints
44state.checkpoints.num-retained: 1
45state.savepoints.dir: hdfs:///flink/flink-savepoints
46
47
48# =============== 禁用flink类加载器，优先使用用户上传===============
49classloader.check-leaked-classloader: false
50classloader.resolve-order: child-first
51

（3）配置workers

1vim workers
2
3node02
4node03
5node04
6

（4）创建必要的目录

1mkdir -p /opt/module/flink/flink-data
2mkdir -p /opt/module/flink/flink-blob
3mkdir -p /opt/module/flink/flink-jar
4mkdir -p /opt/module/flink/flink-zookeeper/data
5mkdir -p /opt/module/flink/flink-zookeeper/logs
6

（5）配置环境变量

1# 编辑环境变量，创建单独文件方便管理
2vim /etc/profile.d/my_env.sh
3
4# JAVA_HOME，JDK文章中已配置，此处不配，仅做展示
5export JAVA_HOME=/opt/module/jdk8
6export PATH=$PATH:$JAVA_HOME/bin
7
8# ZOOKEEPER_HOME，zookeeper文章中已配置，此处不配，仅做展示
9export ZOOKEEPER_HOME=/opt/module/zookeeper
10export PATH=$PATH:$ZOOKEEPER_HOME/bin
11
12# HADOOP_HOME，hadoop文章中已配置，此处不配，仅做展示
13export HADOOP_HOME=/opt/module/hadoop
14export PATH=$PATH:$HADOOP_HOME/bin:$HADOOP_HOME/sbin
15export HADOOP_CLASSPATH=$(hadoop classpath)
16export HADOOP_CONF_DIR=${HADOOP_HOME}/etc/hadoop
17export HADOOP_PID_DIR=/opt/module/hadoop/pid
18export HADOOP_SECURE_PID_DIR=${HADOOP_PID_DIR}
19export HDFS_NAMENODE_USER=root
20export HDFS_DATANODE_USER=root
21export HDFS_JOURNALNODE_USER=root
22export HDFS_ZKFC_USER=root
23export YARN_RESOURCEMANAGER_USER=root
24export YARN_NODEMANAGER_USER=root
25
26# FLINK_HOME
27export FLINK_HOME=/opt/module/flink
28export PATH=$PATH:$FLINK_HOME/bin
29
30# 退出后，刷新环境变量
31source /etc/profile.d/my_env.sh
32

3、分发flink

1# 将配好的flink分发到node03，node04
2scp -r flink/ root@node03:/opt/module/
3scp -r flink/ root@node04:/opt/module/
4

七、集群测试

1、启动zookeeper，hadoop

1# 启动zookeeper，三个节点分别启动
2zkServer.sh start
3
4# 启动hadoop，node02
5# 启动hdfs集群
6start-dfs.sh
7# 启动hdfs集群
8start-yarn.sh
9

2、Yarn Session测试

（1）模式介绍

YARN Session模式特点：

1. 先启动一个长期运行的Flink集群
2. 在该集群上可以提交多个作业
3. 适合短作业频繁提交的场景
4. 资源共享，启动速度快

（2）准备测试资源

1# 创建测试用的HDFS目录
2hdfs dfs -mkdir -p /flink/test/input
3hdfs dfs -mkdir -p /flink/test/output
4
5# 上传测试数据到HDFS
6echo "hello world hello flink
7flink is fast
8hello hadoop
9flink streaming" > test.txt
10
11hdfs dfs -put test.txt /flink/test/input/
12

（3）启动YARN Session

1# 在node02上执行（作为客户端）
2cd $FLINK_HOME
3
4# 方法1：交互式启动（推荐测试用）
5./bin/yarn-session.sh -d
6
7# 方法2：分离模式启动
8./bin/yarn-session.sh -d -jm 1024m -tm 2048m -s 2 -nm "FlinkYarnSession"
9
10# 参数说明：
11# -d: 分离模式（后台运行）
12# -jm: JobManager内存（默认1024m）
13# -tm: 每个TaskManager内存（默认1024m）
14# -s: 每个TaskManager的slot数量（默认1）
15# -nm: YARN应用名称
16# -qu: YARN队列名称
17# -D: 传递Flink配置参数
18
19# 示例：指定更多资源
20./bin/yarn-session.sh \
21  -d \
22  -jm 2048m \
23  -tm 4096m \
24  -s 4 \
25  -nm "FlinkTestSession" \
26  -D taskmanager.memory.network.min=128mb \
27  -D taskmanager.memory.network.max=256mb \
28  -D parallelism.default=4
29

（4）查看YARN Session状态

1# 查看YARN上的应用
2yarn application -list
3
4# 查找Flink Session应用ID
5yarn application -list | grep Flink
6
7# 查看应用详情
8yarn application -status <application_id>
9
10# 也可以直接打开yarn页面查看
11

（5）提交作业

1# 首先找到刚刚启动的session，获取YARN应用ID
2yarn application -list
3
4# 测试1：提交WordCount示例作业
5./bin/flink run \
6  -m yarn-cluster \
7  -yid application_1765702106723_0004 \
8  ./examples/streaming/WordCount.jar \
9  --input hdfs://ns/flink/test/input/test.txt \
10  --output hdfs://ns/flink/test/output/wordcount_result
11
12# 测试2：提交Socket作业（需要先启动netcat）
13# 在一个终端启动netcat
14nc -lk 9999
15
16# 在另一个终端提交Socket作业
17./bin/flink run \
18  -m yarn-cluster \
19  -yid application_1765702106723_0004 \
20  ./examples/streaming/SocketWindowWordCount.jar \
21  --hostname node02 \
22  --port 9999
23

（6）停止YARN Session

1# 方法1：通过YARN命令停止
2yarn application -kill application_1765702106723_0004
3
4# 方法2：通过Flink命令停止
5echo "stop" | ./bin/yarn-session.sh -id application_1765702106723_0004
6
7# 方法3：在Web UI中停止
8# 访问ApplicationMaster Web UI -> Cluster -> Stop Cluster
9

3、Per-Job测试

（1）Per-Job模式介绍

Per-Job模式特点：

1. 每个作业独立启动一个Flink集群
2. 作业完成后集群自动释放资源
3. 资源隔离性好
4. 适合长时间运行的生产作业

（2）提交per-job作业

1./bin/flink run -d \
2  -m yarn-cluster \
3  -ynm "FlinkPerJobWordCount" \
4  -yjm 1024m \
5  -ytm 1024m \
6  -ys 1 \
7  -p 1 \
8  ./examples/batch/WordCount.jar \
9  --input hdfs://ns/flink/test/input/test.txt \
10  --output hdfs://ns/flink/test/output/perjob_wordcount
11  
12  
13nc -lk 9999
14# 在另一个终端提交Socket作业
15bin/flink run -d \
16  -m yarn-cluster \
17  -ynm "FlinkPerJobWordCount" \
18  -yjm 1024m \
19  -ytm 1024m \
20  -ys 1 \
21  -p 1 \
22  ./examples/streaming/SocketWindowWordCount.jar \
23  --hostname node02 \
24  --port 9999
25  
26
27# 参数详细说明：
28# -m yarn-cluster: 指定YARN Per-Job模式
29# -ynm: YARN应用名称
30# -yjm: JobManager内存
31# -ytm: TaskManager内存
32# -ys: 每个TaskManager的slot数量
33# -yqu: YARN队列名称
34# -p: 作业并行度
35# -yD: 传递动态参数，如：-yD taskmanager.memory.network.min=128mb
36

《Linux软件安装 —— Flink集群安装（集成Zookeeper、Hadoop高可用）》 是转载文章，点击查看原文。