Flink1.16.0集群安装

一、集群规划

部署节点	部署内容	节点 IP
k8s-master	jdk11、flink1.16.0	10.52.0.143
k8s-node1	jdk11、flink1.16.0	10.52.0.144
k8s-node2	jdk11、flink1.16.0	10.52.0.145

二、安装前的准备

1、关闭Selinux、防火墙和THP，三台服务器都要操作一次

关闭防火墙
systemctl stop firewalld

永久关闭防火墙
systemctl disable firewalld

sed -i ‘s/^SELINUX=enforcing$/SELINUX=disabled/’ /etc/selinux/config && setenforce 0

2、配置hosts，三台机器都要操作一次

vim /etc/hosts
新增
10.52.0.143 k8s-master
10.52.0.144 k8s-node1
10.52.0.145 k8s-node2

3、三台机器配置root账号免密钥

(1) 生成密钥,三台机器都执行
ssh-keygen
(2)公钥复制
比如在k8s-master上执行完ssh-keygen后，执行
ssh-copy-id root@k8s-node1
ssh-copy-id root@k8s-node2

在k8s-node1上执行：
ssh-copy-id root@k8s-node2
ssh-copy-id root@k8s-master

在k8s-node2上执行：
ssh-copy-id root@k8s-node1
ssh-copy-id root@k8s-master

（3）自己ssh自己
在每台机器上分别执行：
cd ~/.ssh/
cat id_rsa.pub >> authorized_keys

4、Flink版本的选择

1.Flink版本的选择，附Flink官网的地址，https://www.apache.org/dyn/closer.lua/flink/flink-1.16.0/flink-1.16.0-bin-scala_2.12.tgz，
https://dlcdn.apache.org/flink/flink-1.16.0/flink-1.16.0-bin-scala_2.12.tgz

2、Java版本的选择,附zulu版本的jdk地址，https://cdn.azul.com/zulu/bin/，zulu11.56.19-ca-jdk11.0.15-linux_x64.tar.gz

三、部署过程

1、部署Java11

1. 上传zulu11.56.19-ca-jdk11.0.15-linux_x64.tar.gz 到 /opt/modules/software
   解压tar -zxvf zulu11.56.19-ca-jdk11.0.15-linux_x64.tar.gz -C /opt/modules/
   解压完在/opt/modules/下出现 zulu11.56.19-ca-jdk11.0.15-linux_x64

2. 修改名称
   mv /opt/modules/zulu11.56.19-ca-jdk11.0.15-linux_x64 /opt/modules/jdk11

3. 配置scala的环境变量
   vim /etc/profile

   添加：

   export JAVA_HOME=/opt/modules/jdk11
   export PATH=$PATH:$JAVA_HOME/bin
   

   使环境变量生效
   source /etc/profile

   注意：如果机器中本身就装了其他版本的jdk则不需要再配jdk11的环境的变量，等下在flink中直接配就可以了

2、部署Flink

1. 上传Flink压缩包flink-1.16.0-bin-scala_2.12.tgz，并解压到指定位置
   tar -zxvf flink-1.16.0-bin-scala_2.12.tgz -C /opt/modules/
   解压完在/opt/modules/下出现flink-1.16.0

2. 修改名称
   mv /opt/modules/flink-1.16.0 /opt/modules/flink

3. 配置Flink的环境变量
   vim /etc/profile
   添加：

export FLINK_HOME=/opt/modules/flink
export PATH=$PATH:$FLINK_HOME/bin

使环境变量生效
source /etc/profile

4. 解压后进入主配置文件：
   vim flink-conf.yaml

   需要修改：
   如果机器中已经有其他版本的jdk，则需要在flink的配置文件中配置jdk11的路径

   env.java.home: /opt/modules/jdk11

主节点的ip或主机名（前提：/etc/hosts已配置好）
jobmanager.rpc.address: k8s-master
jobmanager.bind-host: 0.0.0.0
如果ssh端口号不是22，需要添加这一行（-p后是真实的远程ssh端口）
env.ssh.opts: -p 9652

每个JobManager的可用内存大小，默认为1024M
jobmanager.memory.process.size: 4096m

每个TaskManager的可用内存大小，默认为1024M
taskmanager.memory.process.size: 8192m

每台计算机可用的CPU数，默认为1
taskmanager.numberOfTaskSlots: 8

集群中默认的并行度
parallelism.default: 8

修改rest地址
rest.address: k8s-master
rest.bind-address: 0.0.0.0

5. 添加主节点信息：
   vim masters
   添加:
   k8s-master:8081

6. 添加从节点的信息：
   vim workers
   添加:
   k8s-master
   k8s-node1
   k8s-node2

7. 将修改好的配置发送到其他两个从节点
   scp -r /opt/modules/flink root@k8s-node1:/opt/modules/
   scp -r /opt/modules/flink root@k8s-node2:/opt/modules/

8. 从节点也和主节点一样如上配置环境变量
   vim /etc/profile
   添加：

export FLINK_HOME=/opt/modules/flink
export PATH=$PATH:$FLINK_HOME/bin

使环境变量生效
source /etc/profile

9. 启动验证
   cd /opt/modules/flink/
   执行./bin/start-cluster.sh

登录Web页面查看
http://k8s-master:8081/

页面展示
首页

看到Available Task Slots显示为24,因为配置的每台计算机8个slot,一共3台计算机，总共24个slot

Standalone模式在命令行执行：
cd /opt/modules/flink
./bin/flink run ./examples/batch/WordCount.jar
看到有wordcount的结果则表明成功了：

yarn-application模式执行
需要先在Flink集群中配置上hadoop客户端，需要将集群中的core-site.xml hdfs-site.xml yarn-site.xml拷贝到hadoop客户端的 etc/hadoop/下替换掉原的
上传flink/lib/下的所有jar文件到hdfs的/jars/flink/下
hadoop fs -put flink/lib/* /jars/flink/
hadoop fs -put flink/examples/batch/* /jars/apps/

进入flink目录下执行
./bin/flink run-application -t yarn-application -Dyarn.provided.lib.dirs=“hdfs://hadoop1:8020/jars/flink/” hdfs://hadoop1:8020/jars/apps/WordCount.jar

也可以指定参数：
./bin/flink run-application -t yarn-application -Dtaskmanager.numberOfTaskSlots=4 -Dparallelism.default=4 -Dyarn.provided.lib.dirs=“hdfs://hadoop1:8020/jars/flink/” hdfs://hadoop1:8020/jars/apps/WordCount.jar