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

一、节点说明

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）上传并解压

# 加压到安装目录tar-zxvfflink-1.13.6-bin-scala_2.12.tgz-C/opt/module/# 修改名称mvflink-1.13.6/ flink

2、修改配置

（1）配置zookeeper

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

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

（2）配置flink-conf.yaml

# 备份原始文件cp flink-conf.yaml flink-conf.yaml.bak# 编辑flink-conf.yaml文件vim /opt/module/flink/conf/flink-conf.yaml# ==================== 基础配置 ====================# JobManager节点，指定node02为JobManagerjobmanager.rpc.address:node02jobmanager.rpc.port:6123# JobManager堆内存（根据实际内存调整，建议4G以上）jobmanager.memory.process.size:1024m# TaskManager堆内存（根据实际内存调整）taskmanager.memory.process.size:1024m# 每个TaskManager的slot数量（根据CPU核心数调整）taskmanager.numberOfTaskSlots:1# 并行度默认值parallelism.default:1web.tmpdir:/opt/module/flink/flink-jarblob.storage.directory:/opt/module/flink/flink-blobyarn.maximum-failed-containers:200taskmanager.tmp.dirs:/opt/module/flink/flink-dataflink_log_bak_dir:/opt/module/flink/logsflink_log_dir:/opt/module/flink/logs# ==================== 高可用配置 ====================high-availability:zookeeperhigh-availability.zookeeper.quorum:node02:2181,node03:2181,node04:2181# Zookeeper中Flink的根路径high-availability.zookeeper.path.root:/flink# JobManager元数据存储路径（使用HDFS）high-availability.zookeeper.storageDir:hdfs:///flink/recoveryfs.hdfs.hadoopconf:/opt/module/hadoop/etc/hadoopfs.hdfs.hdfssite:/opt/module/hadoop/etc/hadoop/hdfs-site.xml# ==================== 故障恢复、checkpoint ====================restart-strategy:fixed-delayrestart-strategy.fixed-delay.attempts:3restart-strategy.fixed-delay.delay:5 sstate.backend:rocksdbstate.backend.incremental:truestate.checkpoint-storage:filesystemstate.checkpoints.dir:hdfs:///flink/flink-checkpointsstate.checkpoints.num-retained:1state.savepoints.dir:hdfs:///flink/flink-savepoints# =============== 禁用flink类加载器，优先使用用户上传===============classloader.check-leaked-classloader:falseclassloader.resolve-order:child-first

（3）配置workers

vimworkers node02 node03 node04

（4）创建必要的目录

mkdir-p/opt/module/flink/flink-datamkdir-p/opt/module/flink/flink-blobmkdir-p/opt/module/flink/flink-jarmkdir-p/opt/module/flink/flink-zookeeper/datamkdir-p/opt/module/flink/flink-zookeeper/logs

（5）配置环境变量

# 编辑环境变量，创建单独文件方便管理vim/etc/profile.d/my_env.sh# JAVA_HOME，JDK文章中已配置，此处不配，仅做展示exportJAVA_HOME=/opt/module/jdk8exportPATH=$PATH:$JAVA_HOME/bin# ZOOKEEPER_HOME，zookeeper文章中已配置，此处不配，仅做展示exportZOOKEEPER_HOME=/opt/module/zookeeperexportPATH=$PATH:$ZOOKEEPER_HOME/bin# HADOOP_HOME，hadoop文章中已配置，此处不配，仅做展示exportHADOOP_HOME=/opt/module/hadoopexportPATH=$PATH:$HADOOP_HOME/bin:$HADOOP_HOME/sbinexportHADOOP_CLASSPATH=$(hadoop classpath)exportHADOOP_CONF_DIR=${HADOOP_HOME}/etc/hadoopexportHADOOP_PID_DIR=/opt/module/hadoop/pidexportHADOOP_SECURE_PID_DIR=${HADOOP_PID_DIR}exportHDFS_NAMENODE_USER=rootexportHDFS_DATANODE_USER=rootexportHDFS_JOURNALNODE_USER=rootexportHDFS_ZKFC_USER=rootexportYARN_RESOURCEMANAGER_USER=rootexportYARN_NODEMANAGER_USER=root# FLINK_HOMEexportFLINK_HOME=/opt/module/flinkexportPATH=$PATH:$FLINK_HOME/bin# 退出后，刷新环境变量source/etc/profile.d/my_env.sh

3、分发flink

# 将配好的flink分发到node03，node04scp-rflink/ root@node03:/opt/module/scp-rflink/ root@node04:/opt/module/

七、集群测试

1、启动zookeeper，hadoop

# 启动zookeeper，三个节点分别启动zkServer.sh start# 启动hadoop，node02# 启动hdfs集群start-dfs.sh# 启动hdfs集群start-yarn.sh

2、Yarn Session测试

（1）模式介绍

YARN Session模式特点：

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

（2）准备测试资源

# 创建测试用的HDFS目录hdfs dfs-mkdir-p/flink/test/input hdfs dfs-mkdir-p/flink/test/output# 上传测试数据到HDFSecho"hello world hello flink flink is fast hello hadoop flink streaming">test.txt hdfs dfs-puttest.txt /flink/test/input/

（3）启动YARN Session

# 在node02上执行（作为客户端）cd$FLINK_HOME# 方法1：交互式启动（推荐测试用）./bin/yarn-session.sh-d# 方法2：分离模式启动./bin/yarn-session.sh-d-jm1024m-tm2048m-s2-nm"FlinkYarnSession"# 参数说明：# -d: 分离模式（后台运行）# -jm: JobManager内存（默认1024m）# -tm: 每个TaskManager内存（默认1024m）# -s: 每个TaskManager的slot数量（默认1）# -nm: YARN应用名称# -qu: YARN队列名称# -D: 传递Flink配置参数# 示例：指定更多资源./bin/yarn-session.sh\-d\-jm2048m\-tm4096m\-s4\-nm"FlinkTestSession"\-Dtaskmanager.memory.network.min=128mb\-Dtaskmanager.memory.network.max=256mb\-Dparallelism.default=4

（4）查看YARN Session状态

# 查看YARN上的应用yarnapplication-list# 查找Flink Session应用IDyarnapplication-list|grepFlink# 查看应用详情yarnapplication-status<application_id># 也可以直接打开yarn页面查看

（5）提交作业

# 首先找到刚刚启动的session，获取YARN应用IDyarnapplication-list# 测试1：提交WordCount示例作业./bin/flink run\-myarn-cluster\-yidapplication_1765702106723_0004\./examples/streaming/WordCount.jar\--inputhdfs://ns/flink/test/input/test.txt\--outputhdfs://ns/flink/test/output/wordcount_result# 测试2：提交Socket作业（需要先启动netcat）# 在一个终端启动netcatnc-lk9999# 在另一个终端提交Socket作业./bin/flink run\-myarn-cluster\-yidapplication_1765702106723_0004\./examples/streaming/SocketWindowWordCount.jar\--hostnamenode02\--port9999

（6）停止YARN Session

# 方法1：通过YARN命令停止yarnapplication-killapplication_1765702106723_0004# 方法2：通过Flink命令停止echo"stop"|./bin/yarn-session.sh-idapplication_1765702106723_0004# 方法3：在Web UI中停止# 访问ApplicationMaster Web UI -> Cluster -> Stop Cluster

3、Per-Job测试

（1）Per-Job模式介绍

Per-Job模式特点：

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

（2）提交per-job作业

./bin/flink run-d\-myarn-cluster\-ynm"FlinkPerJobWordCount"\-yjm1024m\-ytm1024m\-ys1\-p1\./examples/batch/WordCount.jar\--inputhdfs://ns/flink/test/input/test.txt\--outputhdfs://ns/flink/test/output/perjob_wordcountnc-lk9999# 在另一个终端提交Socket作业bin/flink run-d\-myarn-cluster\-ynm"FlinkPerJobWordCount"\-yjm1024m\-ytm1024m\-ys1\-p1\./examples/streaming/SocketWindowWordCount.jar\--hostnamenode02\--port9999# 参数详细说明：# -m yarn-cluster: 指定YARN Per-Job模式# -ynm: YARN应用名称# -yjm: JobManager内存# -ytm: TaskManager内存# -ys: 每个TaskManager的slot数量# -yqu: YARN队列名称# -p: 作业并行度# -yD: 传递动态参数，如：-yD taskmanager.memory.network.min=128mb