Cenots上Flink Standalone 集群部署

环境说明
Centos7.5
flink-1.20.0-bin-scala_2.12.tgz

jdk1.8
hadoop-2.7.3
Flink运行依赖Java环境，如果没有Java环境，可自行下载Java后，再进行下面操作

一，Flink下载解压运行
去Flink官网进行下载，Downloads | Apache Flink，点击链接跳转后，可以选择自己需要的版本进行下载，这里我选择的是flink-1.20.0-bin-scala_2.12.tgz。
下载完成后，上传至虚拟机 /opt 目录下，然后进行解压，解压至/usr/local/下

cd /opt
tar -xvf flink-1.20.0-bin-scala_2.12.tgz -C /usr/local

就会在/usr/local下生成新文件夹“flink-1.20.0”

然后执行以下命令就可以启动单机版本的 Flink：

cd /usr/local/flink-1.20.0
bin/start-cluster.sh

启动后，查看jps，如下图，包括一个taskmanager和一个ClusterEntryPoint进程

jps

二，Web UI界面

Flink 提供了 WEB 界面用于直观的管理 Flink 集群，访问端口为 8081

通过访问，主机IP＋端口号即可访问，例如我的为：192.168.74.110:8081

如果访问不通，解决方法如下
此时需要修改Flink配置文件conf/config.yaml文件中的rest.bind-address属性为“0.0.0.0”.

vi config.yaml

修改完成之后，重启Flink服务就可以访问了

三，测试实例

启动后可以运行安装包中自带的词频统计案例，具体步骤如下：
1）开启端口

nc -lk 9999

输入命令后，命令行会卡住，这是正常现象，这是因为

nc

命令正在等待接收来自网络的数据。
2）运行实例
新打开一个命令行窗口，然后跳转至Flink目录下

cd /usr/local/flink-1.20.0

然后执行以下命令，运行实例

bin/flink run examples/streaming/SocketWindowWordCount.jar --port 9999

就会出现

3）输入测试数据
此时我们跳转回，我们刚才开启端口的那个命令窗口

输入测试数据,中间有空格

a a b b c c d d a b

4）查看控制台输出
打开WEB UI页面，首先看到我们有一个实例即作业在运行中

然后依次点击下面的操作，就可以看到实例运行结果

5）停止作业（实例）
①方式一，通过Web UI 页面的Cancel Job按钮
从左到右依次点击

然后进入页面后，点击右上角，结束作业

**②方式二使用

flink cancel

命令取消作业：**
刚才在启动后，得到了JobId，如图

此时我们就可以打开新的命令行窗口输入以下命令

cd /usr/local/flink-1.20.0
bin/flink cancel "你的jobid"

不知道JobId怎么办，只需同样打开新的命令行窗口
输入以下命令

cd /usr/local/flink-1.20.0
./bin/flink list

然后你就会看到某个作业的JobId了

然后再执行bin/flink命令＋Jobid即可
6）停止Flink

bin/stop-cluster.sh

四，Standalone Cluster

Standalone Cluster 模式是 Flink 自带的一种集群模式，具体配置步骤如下：

1）提前准备
四台服务器，主机名分别为hadoop01 hadoop02 hadoop03 hadoop04，其中hadoop01为主节点
其余三台为slave节点

2）搭建步骤
修改config.yaml 中 jobmanager 节点的通讯地址为 hadoop01，bind-host为: 0.0.0.0

cd /usr/local/flink-1.20.0/conf
vi config.yaml

修改 workers配置文件，将 hadoop02 和 hadoop03和hadoop04 配置为 slave 节点：

vi workers

删除localhost
加入

hadoop02
hadoop03
hadoop04

将配置好的 Flink 安装包分发到其他两台服务器上：执行以下命令

 scp -r /usr/local/flink-1.20.0 hadoop02:/usr/local
 scp -r /usr/local/flink-1.20.0 hadoop03:/usr/local
 scp -r /usr/local/flink-1.20.0 hadoop04:/usr/local

完成后，输入以下命令在hadoop01上启动集群(前面那个作业记得停止，要不然这里会有警告)

cd /usr/local/flink-1.20.0
bin/start-cluster.sh

启动完毕后，你应该可以在主节点输入jps可以看到

各子节点可以看到

Web UI界面可以看到

五，Flink Standalone HA搭建
Standalone Cluster HA（High Availability）是Flink中的一种高可用性配置，用于确保Flink集群在发生故障时能够继续运行。
1）提前准备

Hadoop 集群和 ZooKeeper 集群的部署搭建
2）搭建步骤
①修改 config.yaml 文件

cd /usr/local/flink-1.20.0/conf
vi config.yaml

增加如下配置：注意自己hadoop端口号和主机名

# 配置使用zookeeper来开启高可用模式
high-availability: zookeeper
# 配置zookeeper的地址，采用zookeeper集群时，可以使用逗号来分隔多个节点地址
high-availability.zookeeper.quorum: hadoop01:2181,hadoop02:2181,hadoop03:2181,hadoop04:2181
# 在zookeeper上存储flink集群元信息的路径
high-availability.zookeeper.path.root: /flink
# 集群id
high-availability.cluster-id: /standalone_cluster_one
# 持久化存储JobManager元数据的地址，zookeeper上存储的只是指向该元数据的指针信息
high-availability.storageDir: hdfs://hadoop01:9000/flink/recovery

修改conf/masters文件，修改内容如下：

hadoop01:8081
hadoop02:8081
hadoop03:8081

这代表将 hadoop01 和 hadoop02 和hadoop03都配置为 master 节点：
修改conf/zoo.cfg文件
在里面加入以下代码

server.1=hadoop01:2888:3888
server.2=hadoop02:2888:3888
server.3=hadoop03:2888:3888

这个文件里最下面本来有server.1=localhost:2888:3888，删除掉这句话
②再将这些配置复制到其他节点

 scp -r /usr/local/flink-1.20.0 hadoop02:/usr/local
 scp -r /usr/local/flink-1.20.0 hadoop03:/usr/local
 scp -r /usr/local/flink-1.20.0 hadoop04:/usr/local

③启动ZooKeeper集群
在主节点Flink安装目录下运行