flink on yarn HA部署
登录
侧边栏壁纸
博主昵称
clouds

  • 累计撰写 18 篇文章
  • 累计收到 15 条评论
Linux

flink on yarn HA部署

zy2025
zy2025
2025-05-29 / 1 评论 / 4 阅读 / 正在检测是否收录...

1.准备工作目录

[root@flink111 ~]# mkdir /zy/{data/{flink,hadoop,zookeeper},logs/{flink,hadoop,zookeeper},app} -p
[root@flink112 ~]# mkdir /zy/{data/{flink,hadoop,zookeeper},logs/{flink,hadoop,zookeeper},app} -p
[root@flink113 ~]# mkdir /zy/{data/{flink,hadoop,zookeeper},logs/{flink,hadoop,zookeeper},app} -p

2.准备安装包传到/zy/app目录

[root@flink111 ~]# cd /zy/app/
[root@flink111 app]# ll
total 743324
-rw-r--r-- 1 root root 314015504 Jun 11 18:44 flink-1.13.6-bin-scala_2.11.tgz
-rw-r--r-- 1 root root 218720521 Jun 13 16:03 hadoop-2.7.7.tar.gz
-rw-r--r-- 1 root root 191753373 Jun 11 18:43 jdk-8u191-linux-x64.tar.gz
-rw-r--r-- 1 root root  36667596 Jun 13 15:18 zookeeper-3.4.12.tar.gz
-rw-r--r-- 1 root root  41368997 Jul 12 22:06 flink-shaded-hadoop-2-uber-2.7.5-10.0.jar  #flink的依赖包

3.配置集群间免密钥登陆

1.在每个节点执行如下命令，然后一直回车即可
[root@flink111 ~]# ssh-keygen -t rsa
[root@flink112 ~]# ssh-keygen -t rsa
[root@flink113 ~]# ssh-keygen -t rsa

2.分别将从节点的/root/.ssh/中的id_rsa.pub文件拷贝至主节点对应位置（需要重命名已避免冲突）
[root@flink112 .ssh]# scp id_rsa.pub 10.0.0.111:~/.ssh/id_rsa.pub.flink-112
[root@flink113 .ssh]# scp id_rsa.pub 10.0.0.111:~/.ssh/id_rsa.pub.flink-113

3.将主节点中的id_rsa.pub⽂件添加到authorized_keys⽂件中
[root@flink111 .ssh]# cat id_rsa.pub >> authorized_keys
[root@flink111 .ssh]# cat id_rsa.pub.flink-112 >> authorized_keys
[root@flink111 .ssh]# cat id_rsa.pub.flink-113 >> authorized_keys

4.将authorized_keys⽂件拷⻉⾄其他节点即可
[root@flink111 .ssh]# scp authorized_keys 10.0.0.112:~/.ssh/
[root@flink111 .ssh]# scp authorized_keys 10.0.0.113:~/.ssh/

5.然后从1号节点开始，分别ssh其他所有节点。⽬的是为了记住机器编号。也就是known_host
[root@flink111 .ssh]# ssh 10.0.0.111
[root@flink111 .ssh]# ssh 10.0.0.112
[root@flink111 .ssh]# ssh 10.0.0.113

[root@flink112 .ssh]# ssh 10.0.0.111
[root@flink112 .ssh]# ssh 10.0.0.112
[root@flink112 .ssh]# ssh 10.0.0.113

[root@flink113 .ssh]# ssh 10.0.0.111
[root@flink113 .ssh]# ssh 10.0.0.112
[root@flink113 .ssh]# ssh 10.0.0.113

6.查看authorized_keys⽂件，如果⽂件⽤户组权限为读和写，去掉⽤户组写权限即可。
[root@flink111 .ssh]# chmod g-w authorized_keys
[root@flink112 .ssh]# chmod g-w authorized_keys
[root@flink113 .ssh]# chmod g-w authorized_keys

4.安装JDK

1.解压软件包
[root@flink111 app]# tar xf jdk-8u191-linux-x64.tar.gz
[root@flink111 app]# tar xf zookeeper-3.4.12.tar.gz 
[root@flink111 app]# tar xf flink-1.13.6-bin-scala_2.11.tgz 
[root@flink111 app]# tar xf hadoop-2.7.7.tar.gz

5.配置环境变量

[root@flink111 app]# vim /etc/profile
#在最下面增加以下内容:
export JAVA_HOME=/zy/app/jdk1.8.0_191
export CLASSPATH=$JAVA_HOME/lib:$JRE_HOME/:lib:$CLASSPATH
export ZOOKEEPER_HOME=/zy/app/zookeeper-3.4.12
export HADOOP_HOME=/zy/app/hadoop-2.7.7
export FLINK_HOME=/zy/app/flink-1.13.6
export PATH=$PATH:$JAVA_HOME/bin:$ZOOKEEPER_HOME/bin:$HADOOP_HOME/sbin:$HADOOP_HOME/bin:$FLINK_HOME/bin
[root@flink111 app]# source /etc/profile

6.安装zookeeper

1.修改配置文件zoo.cfg
[root@flink111 ~]# cp /zy/app/zookeeper-3.4.12/conf/{zoo_sample,zoo}.cfg
[root@flink111 ~]# vim /zy/app/zookeeper-3.4.12/conf/zoo.cfg
...
# 指定数据目录
dataDir=/zy/data/zookeeper
#指定日志目录
dataLogDir=/zy/logs/zookeeper
...
server.0=10.0.0.111:2888:3888
server.1=10.0.0.112:2888:3888
server.2=10.0.0.113:2888:3888

2.将配置文件同步到其他节点
[root@flink111 ~]# scp /zy/app/zookeeper-3.4.12/conf/zoo.cfg 10.0.0.112:/zy/app/zookeeper-3.4.12/conf/
[root@flink111 ~]# scp /zy/app/zookeeper-3.4.12/conf/zoo.cfg 10.0.0.113:/zy/app/zookeeper-3.4.12/conf/

3.各节点创建myid文件
[root@flink111 ~]# cd /zy/data/zookeeper/
[root@flink111 zookeeper]# echo 0 > myid

[root@flink112 ~]# cd /zy/data/zookeeper/
[root@flink112 zookeeper]# echo 1 > myid

[root@flink113 ~]# cd /zy/data/zookeeper/
[root@flink113 zookeeper]# echo 2 > myid

3.各节点启动zookeeper
[root@flink111 zookeeper]# zkServer.sh start
[root@flink112 zookeeper]# zkServer.sh start
[root@flink113 zookeeper]# zkServer.sh start

7.安装hadoop

1.各节点准备工作目录
[root@flink111 ~]# mkdir -p /zy/data/hadoop/{tmp,journal,hdfs/{name,data,namesecondary},nm-local-dir,pid}
[root@flink112 ~]# mkdir -p /zy/data/hadoop/{tmp,journal,hdfs/{name,data,namesecondary},nm-local-dir,pid}
[root@flink113 ~]# mkdir -p /zy/data/hadoop/{tmp,journal,hdfs/{name,data,namesecondary},nm-local-dir,pid}

2.修改配置文件
(1) core-site.xml
[root@flink111 ~]# cd /zy/app/hadoop-2.7.7/etc/hadoop
[root@flink111 hadoop]# cat core-site.xml
#在这里面增加以下内容
<configuration>
    <!-- 指定hdfs的nameservice为ns -->
    <property>
     <name>fs.defaultFS</name>
     <value>hdfs://ns</value>
    </property>
    <!--指定hadoop数据临时存放⽬录-->
    <property>
     <name>hadoop.tmp.dir</name>
     <value>/zy/data/hadoop/tmp</value>
    </property>
    <property>
     <name>io.file.buffer.size</name>
     <value>4096</value>
    </property>
    <!--指定zookeeper地址-->
    <property>
     <name>ha.zookeeper.quorum</name>
     <value>flink111:2181,flink112:2181,flink113:2181</value>
    </property>
</configuration>

(2) hdfs-site.xml
[root@flink111 hadoop]# cat hdfs-site.xml
...
<configuration>
    <!--指定hdfs的nameservice为ns，需要和core-site.xml中的保持⼀致 -->
     <property>
     <name>dfs.nameservices</name>
     <value>ns</value>
     </property>
     <!-- ns下⾯有两个NameNode，分别是nn1，nn2 -->
     <property>
     <name>dfs.ha.namenodes.ns</name>
     <value>nn1,nn2</value>
     </property>
     <!-- nn1的RPC通信地址 -->
     <property>
     <name>dfs.namenode.rpc-address.ns.nn1</name>
     <value>flink111:9000</value>
     </property>
     <!-- nn1的http通信地址 -->
     <property>
     <name>dfs.namenode.http-address.ns.nn1</name>
     <value>flink111:50070</value>
     </property>
     <!-- nn2的RPC通信地址 -->
     <property>
     <name>dfs.namenode.rpc-address.ns.nn2</name>
     <value>flink112:9000</value>
     </property>
     <!-- nn2的http通信地址 -->
     <property>
     <name>dfs.namenode.http-address.ns.nn2</name>
     <value>flink112:50070</value>
     </property>
    <!-- 指定NameNode的元数据在JournalNode上的存放位置 -->
     <property>
     <name>dfs.namenode.shared.edits.dir</name>
     <value>qjournal://flink111:8485;flink112:8485;flink113:8485/ns</value>
     </property>
     <!-- 指定JournalNode在本地磁盘存放数据的位置 -->
     <property>
     <name>dfs.journalnode.edits.dir</name>
     <value>/zy/data/hadoop/journal</value>
     </property>
     <!-- 开启NameNode故障时⾃动切换 -->
     <property>
     <name>dfs.ha.automatic-failover.enabled</name>
     <value>true</value>
     </property>
     <!-- 配置失败⾃动切换实现⽅式 -->
     <property>
     <name>dfs.client.failover.proxy.provider.ns</name>
     <value>org.apache.hadoop.hdfs.server.namenode.ha.ConfiguredFailoverProxyProvider</value>
     </property>
     <!-- 配置隔离机制，如果ssh是默认22端⼝，value直接写sshfence即可 -->
     <property>
     <name>dfs.ha.fencing.methods</name>
     <value>sshfence</value>
     </property>
     <!-- 使⽤隔离机制时需要ssh免登陆 -->
     <property>
     <name>dfs.ha.fencing.ssh.private-key-files</name>
     <value>/root/.ssh/id_rsa</value>
     </property>
     <property>
     <name>dfs.namenode.name.dir</name>
     <value>file:/zy/data/hadoop/hdfs/name</value>
     </property>
     <property>
     <name>dfs.datanode.data.dir</name>
     <value>file:/zy/data/hadoop/hdfs/data</value>
     </property>
     <property>
     <name>dfs.replication</name>
     <value>2</value>
     </property>
     <!-- 在NN和DN上开启WebHDFS (REST API)功能,不是必须 -->
     <!--webui页面可以操作文件-->
     <property>
     <name>dfs.webhdfs.enabled</name>
     <value>true</value>
     </property> 
</configuration>

(3) yarn-site.xml
[root@flink111 hadoop]# cat yarn-site.xml
...
<configuration>
     <property>
     <name>yarn.nodemanager.aux-services</name>
     <value>mapreduce_shuffle</value>
     </property>
     <property>
     <name>yarn.log-aggregation-enable</name>
     <value>true</value>
     </property>
     <property>
     <name>yarn.log-aggregation.retain-seconds</name>
     <value>604800</value>
     </property>
     <property>
     <name>yarn.resourcemanager.ha.enabled</name>
     <value>true</value>
     </property>
     <property>
     <name>yarn.resourcemanager.cluster-id</name>
     <value>clusterrm</value>
     </property>
     <property>
     <name>yarn.resourcemanager.ha.rm-ids</name>
     <value>rm1,rm2</value>
     </property>
     <property>
     <name>yarn.resourcemanager.hostname.rm1</name>
     <value>flink111</value>
     </property>
     <property>
     <name>yarn.resourcemanager.hostname.rm2</name>
     <value>flink112</value>
     </property>          
     <property>
     <name>yarn.resourcemanager.zk-address</name>
     <value>flink111:2181,flink112:2181,flink113:2181</value>
     </property>
     <property>
     <name>yarn.resourcemanager.recovery.enabled</name>
     <value>true</value>
     </property>
     <property>
     <name>yarn.resourcemanager.store.class</name>
     <value>org.apache.hadoop.yarn.server.resourcemanager.recovery.ZKRMStateStore</value>
     </property>
     <property>
     <name>yarn.scheduler.maximum-allocation-mb</name>
     <value>8196</value>
     </property>
     <property>
     <name>yarn.scheduler.minimum-allocation-mb</name>
     <value>1024</value>
     </property>
     <property>
     <name>yarn.nodemanager.resource.memory-mb</name>
     <value>56320</value>
     </property>
     <property>
     <name>yarn.nodemanager.resource.cpu-vcores</name>
     <value>64</value>
     </property>
     <property>
     <name>yarn.nodemanager.vmem-pmem-ratio</name>
     <value>4.2</value>
     </property>
     <property>
     <name>yarn.resourcemanager.scheduler.class</name>
     <value>org.apache.hadoop.yarn.server.resourcemanager.scheduler.fair.FairScheduler</value>
     </property>
     <property>
     <name>yarn.scheduler.fair.preemption</name>
     <value>true</value>
     </property>
    <!-- 下⾯配置⽤来设置集群利⽤率的阀值， 默认值0.8f，最多可以抢占到集群所有资源的80% -->
     <property>
     <name>yarn.scheduler.fair.preemption.cluster-utilization-threshold</name>
     <value>1.0</value>
     </property>
     <property>
     <name>yarn.nodemanager.local-dirs</name>
     <value>/zy/data/hadoop/nm-local-dir</value>
     </property>
</configuration>

(4) hadoop-env.sh
[root@flink111 hadoop]# cat hadoop-env.sh
#在最下面增加以下内容
export JAVA_HOME=/zy/app/jdk1.8.0_191
export HADOOP_PID_DIR=/zy/data/hadoop/pid
export HADOOP_SECURE_DN_PID_DIR=${HADOOP_PID_DIR}  #新版本HADOOP_SECURE_PID_DIR=${HADOOP_PID_DIR} 

(5) yarn-env.sh
[root@flink111 hadoop]# cat yarn-env.sh
#在最下面增加以下内容
export JAVA_HOME=/zy/app/jdk1.8.0_191
export YARN_PID_DIR=/zy/data/hadoop/pid           #新版本HADOOP_PID_DIR=/zy/data/hadoop/pid

(6) mapred-site.xml
[root@flink111 hadoop]# cp mapred-site.xml.template mapred-site.xml
[root@flink111 hadoop]# cat mapred-site.xml
#在这里面增加以下内容
<configuration>
     <property>
     <name>mapreduce.framework.name</name>
     <value>yarn</value>
    </property>
</configuration>

(7) workers
[root@flink111 hadoop]#  cat workers
localhost
#集群主机的名称:
flink111
flink112
flink113

(8) /etc/hosts
[root@flink111 hadoop]# cat /etc/hosts
#在最下面增加以下内容
10.0.0.111 flink111
10.0.0.112 flink112
10.0.0.113 flink113

3.将工作目录和配置文件传到各节点
[root@flink111 hadoop]# scp -r /zy/app/hadoop-2.7.7 10.0.0.112:/zy/app/
[root@flink111 hadoop]# scp -r /zy/app/hadoop-2.7.7 10.0.0.113:/zy/app/
[root@flink111 hadoop]# scp /etc/hosts 10.0.0.112:/etc/hosts
[root@flink111 hadoop]# scp /etc/hosts 10.0.0.113:/etc/hosts

8.安装flink

1.修改配置文件
(1) flink-conf.yaml
[root@flink111 ~]# cd /zy/app/flink-1.13.6/conf
[root@flink111 conf]# cp flink-conf.yaml flink-conf.yaml_bak
[root@flink111 conf]# vim flink-conf.yaml
...
jobmanager.rpc.address: flink111
#将原本的内容注释掉
#jobmanager.memory.process.size: 1600m
#taskmanager.memory.process.size: 1728m
#taskmanager.numberOfTaskSlots: 1
#在最下面增加以下内容:
web.tmpdir: /zy/data/flink/flink-jar
blob.storage.directory: /zy/data/flink/flink-blob
yarn.maximum-failed-containers: 20000
taskmanager.tmp.dirs: /zy/data/flink/flink-data
fs.hdfs.hadoopconf: /zy/app/hadoop-2.7.7/etc/hadoop/
fs.hdfs.hdfssite: /zy/app/hadoop-2.7.7/etc/hadoop/hdfs-site.xml
high-availability: zookeeper
high-availability.zookeeper.quorum: flink111:2182,flink112:2182,flink113:2182
high-availability.zookeeper.storageDir: hdfs:///flink/recovery
high-availability.zookeeper.path.root: /flink

jobmanager.memory.process.size: 6000m
jobmanager.heap.mb: 2500
jobmanager.memory.off-heap.size: 1000m
jobmanager.memory.jvm-metaspace.size: 1500m
taskmanager.memory.task.heap.size: 2500m
taskmanager.memory.framework.heap.size: 256m
taskmanager.memory.managed.size: 200m
taskmanager.memory.task.off-heap.size: 1000m
taskmanager.memory.process.size: 6000m
taskmanager.memory.jvm-metaspace.size: 800m
taskmanager.numberOfTaskSlots: 12
jobmanager.archive.fs.dir: hdfs:///flinkHistory/completed-jobs/
historyserver.web.address: 0.0.0.0
historyserver.web.port: 49999
historyserver.archive.fs.dir: hdfs:///flinkHistory/completed-jobs/

(2) zoo.cfg
[root@flink111 conf]# vim zoo.cfg
#修改一下内容:
dataDir=/zy/data/flink/flink-zookeeper-data
clientPort=2182
server.0=10.0.0.111:2889:3889
server.1=10.0.0.112:2889:3889
server.2=10.0.0.113:2889:3889

(3) workers 或者 slaves #根据版本的不同，这里有的是workers，有的是slaves
[root@flink111 conf]# cat workers 
localhost
#集群主机的名称:
flink111
flink112
flink113

2.将依赖包拷⻉⾄lib⽬录下
[root@flink111 conf]# cp /zy/app/flink-shaded-hadoop-2-uber-2.7.5-10.0.jar /zy/app/flink-1.13.6/lib/

3.各节点创建flink data⽬录
[root@flink111 ~]# mkdir -p /zy/data/flink/{flink-data,flink-zookeeper-data,flink-jar}
[root@flink112 ~]# mkdir -p /zy/data/flink/{flink-data,flink-zookeeper-data,flink-jar}
[root@flink113 ~]# mkdir -p /zy/data/flink/{flink-data,flink-zookeeper-data,flink-jar}

4.将flink与data⽂件夹分发到从节点中
[root@flink111 conf]# scp -r /zy/app/flink-1.13.6 10.0.0.112:/zy/app/
[root@flink111 conf]# scp -r /zy/app/flink-1.13.6 10.0.0.113:/zy/app/
[root@flink111 conf]# scp -r /zy/data/flink/* 10.0.0.113:/zy/data/flink/
[root@flink111 conf]# scp -r /zy/data/flink/* 10.0.0.112:/zy/data/flink/

5.启动环境
(1) 启动zookeeper，在三个节点执⾏如下命令 #已启动就跳过此步
[root@flink111 ~]# zkServer.sh start
[root@flink112 ~]# zkServer.sh start
[root@flink113 ~]# zkServer.sh start

[root@flink111 ~]# zkServer.sh status
[root@flink112 ~]# zkServer.sh status
[root@flink113 ~]# zkServer.sh status

(2)（首次启动）创建命名空间，在namenode主节点（flink111）执⾏如下命令。
[root@flink111 ~]# hdfs zkfc -formatZK

(3) 启动journalnode，在三个节点（hdfs-site.xml journalnode配置一致）执⾏如下命令
[root@flink111 ~]# hadoop-daemon.sh start journalnode #较新版本启动hdfs --daemon start journalnode  
[root@flink112 ~]# hadoop-daemon.sh start journalnode
[root@flink113 ~]# hadoop-daemon.sh start journalnode

(4)（首次启动）格式化namenode与journalnode⽬录，在namenode主节点（flink111）执⾏如下命令。
[root@flink111 ~]# hdfs namenode -format ns

(5) 启动namenode进程，在namenode主节点（flink111）执⾏如下命令
[root@flink111 ~]# hadoop-daemon.sh start namenode  #较新版本启动hdfs --daemon start namenode

(6)（首次启动）将元数据从namenode主节点（flink111）拷⻉⾄namenode备节点（flink112）并格式化
在namenode备节点(flink112)执⾏如下命令
[root@flink112 ~]# hdfs namenode -bootstrapStandby

(7) 启动namenode进程，在namenode备节点(flink112)执⾏如下命令
[root@flink112 ~]# hadoop-daemon.sh start namenode  #较新版本启动hdfs --daemon start namenode

(8) 在两个namenode节点都执⾏以下命令
[root@flink111 ~]# hadoop-daemon.sh start zkfc #较新版本启动hdfs --daemon start zkfc
[root@flink112 ~]# hadoop-daemon.sh start zkfc #较新版本启动hdfs --daemon start zkfc

(9) 启动datanode，在各个节点执⾏如下命令
[root@flink111 ~]# hadoop-daemon.sh start datanode #较新版本启动hdfs --daemon start datanode
[root@flink112 ~]# hadoop-daemon.sh start datanode 
[root@flink113 ~]# hadoop-daemon.sh start datanode 

(10) 启动resourcemanager，在主节点（flink111）和备节点（flink112）执⾏如下命令
[root@flink111 ~]# yarn-daemon.sh start resourcemanager #较新版本yarn --daemon start resourcemanager
[root@flink112 ~]# yarn-daemon.sh start resourcemanager

(11) 启动nodemanager，在各个节点执⾏如下命令
[root@flink111 ~]# yarn-daemon.sh start nodemanager     #较新版本yarn --daemon start nodemanager
[root@flink112 ~]# yarn-daemon.sh start nodemanager
[root@flink113 ~]# yarn-daemon.sh start nodemanager

(12) 启动flink，在主节点（flink111）执⾏如下命令
[root@flink111 ~]# start-zookeeper-quorum.sh

6. 配置C:\Windows\System32\drivers\etc\hosts文件
10.0.0.111 flink111
10.0.0.112 flink112
10.0.0.113 flink113

7.在浏览器中键⼊flink111:8088查看yarn application状态

8.停止环境
停止flink，在主节点（flink111）执⾏如下命令
stop-zookeeper-quorum.sh

关闭hadoop，在主节点（flink111）执⾏如下命令
stop-all.sh

关闭zookeeper，各个节点执⾏如下命令
zkServer.sh stop

jps查是否有未关闭的相关进程，有则kill掉

9.新增session
nohup yarn-session.sh -nm loca_test > /zy/data/flink/flink-out/loca_test.out &

10.删除
yarn application -kill application_1675914412308_0332

11.查看
yarn application -list

12.清空所有session
yarn resourcemanager -format-state-store  #主节点执行后会重启resourcemanager，8088端口会转移到备节点。之后备节点执行一次即可转移回原本的。
13.新版本中遇到创建会话失败：
export HADOOP_CLASSPATH=`hadoop classpath`
1
暂无标签
版权属于： zy2025
本文链接： https://clouds.emal.cc/archives/9/
作品采用： 《 署名-非商业性使用-相同方式共享 4.0 国际 (CC BY-NC-SA 4.0) 》许可协议授权

评论 (1)

取消
  1. 头像
    zy2025 作者
    Windows 10 · Google Chrome

    yyds

    回复