2024年最全01-Flink安装部署及入门案例（仅供学习）

/export/server/flink-standalone/examples/batch/WordCount.jar
–input hdfs://node1:8020/wordcount/input/words.txt

![在这里插入图片描述](https://img-blog.csdnimg.cn/direct/3ab1969ae0c54bb68230e85ef32d1755.png#pic_center)



1
2
3
4
5

使用–output指定处理结果数据存储目录

/export/server/flink-standalone/bin/flink run
/export/server/flink-standalone/examples/batch/WordCount.jar
–input hdfs://node1:8020/wordcount/input/words.txt
–output hdfs://node1:8020/wordcount/output/result

[root@node1 ~]# hdfs dfs -text /wordcount/output/result

![在这里插入图片描述](https://img-blog.csdnimg.cn/direct/39244ea7115a4c3fb2b4efd3bbc2e9c9.png#pic_center)  
 5）、关闭Standalone集群服务



1
2
3
4
5
6

一键停止所有服务JobManager和TaskManagers

[root@node1 ~]# /export/server/flink-standalone/bin/stop-cluster.sh
Stopping taskexecutor daemon (pid: 6600) on host node1.
Stopping taskexecutor daemon (pid: 3016) on host node2.
Stopping taskexecutor daemon (pid: 3034) on host node3.
Stopping standalonesession daemon (pid: 6295) on host node1.

**补充**：Flink Standalone集群启动与停止，也可以逐一服务启动



1
2
3
4
5

每个服务单独启动

在node1上启动

/export/server/flink-standalone/bin/jobmanager.sh start

在node1、node2、node3.

/export/server/flink-standalone/bin/taskmanager.sh start # 每台机器执行

===============================================================

每个服务单独停止

在node1上停止

/export/server/flink-standalone/bin/jobmanager.sh stop

在node1、node2、node3

/export/server/flink-standalone/bin/taskmanager.sh stop

#### 07-安装部署之Standalone HA


从Standalone架构图中，可发现JobManager存在`单点故障（SPOF`），一旦JobManager出现意外，整个集群无法工作。为了确保集群的高可用，需要搭建Flink的Standalone HA。  
 ![在这里插入图片描述](https://img-blog.csdnimg.cn/direct/715af74eb954485cb1bc3d5e6e94a68a.png#pic_center)Flink Standalone HA集群，类似YARN HA 集群安装部署，可以启动多个主机点JobManager，使用Zookeeper集群监控JobManagers转态，进行选举leader，实现自动故障转移。  
 ![在这里插入图片描述](https://img-blog.csdnimg.cn/direct/070a87654e934229a1a18131a0c456a7.png#pic_center) 在 Zookeeper 的协助下，一个 Standalone的Flink集群会同时有多个活着的 JobManager，其中\*\*只有一个处于Active工作状态，其他处于 Standby 状态。\*\*当工作中的 JobManager 失去连接后(如宕机或 Crash)，Zookeeper 会从 Standby 中选一个新的 JobManager 来接管 Flink 集群。  
 1）、集群规划  
 ![在这里插入图片描述](https://img-blog.csdnimg.cn/direct/2f7bb0db07d142d08ae056b87b1b4558.png#pic_center)



1
2
3
4
5
6
7
8
9
10
11
12

在node1上复制一份standalone

[root@node1 ~]# cd /export/server/
[root@node1 server]# cp -r flink-standalone flink-ha

删除日志文件

[root@node1 ~]# rm -rf /export/server/flink-ha/log/*

2）、启动ZooKeeper，在node1上启动



1
2
3
4
5

start-zk.sh

3）、启动HDFS，在node1上启动，如果没有关闭，不用重启



1
2
3
4
5

start-dfs.sh

4）、停止集群，在node1操作，进行HA高可用配置



1
2
3
4
5

/export/server/flink-standalone/bin/stop-cluster.sh

5）、修改flink-conf.yaml，在node1操作



1
2
3
4
5

vim /export/server/flink-ha/conf/flink-conf.yaml
修改内容：
jobmanager.rpc.address: node1

high-availability: zookeeper
high-availability.storageDir: hdfs://node1:8020/flink/ha/
high-availability.zookeeper.quorum: node1:2181,node2:2181,node3:2181
high-availability.zookeeper.path.root: /flink
high-availability.cluster-id: /cluster_standalone

state.backend: filesystem
state.backend.fs.checkpointdir: hdfs://node1:8020/flink/checkpoints
state.savepoints.dir: hdfs://node1:8020/flink/savepoints

6）、修改masters，在node1操作



1
2
3
4
5

vim /export/server/flink-ha/conf/masters
修改内容：
node1:8081
node2:8081

7）、分发到集群其他机器，在node1操作



1
2
3
4
5

scp -r /export/server/flink-ha root@node2:/export/server/
scp -r /export/server/flink-ha root@node3:/export/server/

8）、修改node2上的flink-conf.yaml



1
2
3
4
5

[root@node2 ~]# vim /export/server/flink-ha/conf/flink-conf.yaml
修改内容：33 行
jobmanager.rpc.address: node2

9）、重新启动Flink集群



1
2
3
4
5

node1和node2上执行

/export/server/flink-ha/bin/jobmanager.sh start

node1和node2、node3执行

/export/server/flink-ha/bin/taskmanager.sh start # 每台机器执行

![在这里插入图片描述](https://img-blog.csdnimg.cn/direct/d100003f20eb4d6ea53b5c5d7b651e1d.png#pic_center)


#### 08-Flink on YARN之运行流程


​ 在一个企业中，为了最大化的利用集群资源，一般都会在一个集群中同时运行多种类型的Workload，因此 Flink 也支持在 Yarn 集群运行。



> 
> 为什么使用`Flink on Yarn或Spark on Yarn?`
> 
> 
> 


* 1）、Yarn的资源可以按需使用，提高集群的资源利用率
* 2）、Yarn的任务有优先级，根据优先级运行作业
* 3）、基于Yarn调度系统，能够自动化地处理各个角色的 Failover(容错)

 `当应用程序（MR、Spark、Flink）运行在YARN集群上时，可以实现容灾恢复。`


#### 09-Flink on YARN之安装部署


Flink on YARN安装配置，此处考虑高可用HA配置，集群机器安装软件框架示意图：  
 ![在这里插入图片描述](https://img-blog.csdnimg.cn/direct/4085371818cf4166824b579e6f522631.png#pic_center)1）、关闭YARN的内存检查（`node1`操作）



1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33

yarn-site.xml中添加配置

vim /export/server/hadoop/etc/hadoop/yarn-site.xml

添加如下内容：



1
2
3
4
5

yarn.nodemanager.pmem-check-enabled false yarn.nodemanager.vmem-check-enabled false

2）、 配置Application最大的尝试次数（`node1`操作）



1
2
3
4
5

yarn-site.xml中添加配置

vim /export/server/hadoop/etc/hadoop/yarn-site.xml

添加如下内容：



1
2
3
4
5

yarn.resourcemanager.am.max-attempts 4

3）、同步yarn-site.xml配置文件（`node1`操作）



1
2
3
4
5

cd /export/server/hadoop/etc/hadoop
scp -r yarn-site.xml root@node2: P W D s c p − r y a r n − s i t e . x m l r o o t @ n o d e 3