Flink简介、快速入门、部署、集群

文章目录

一、Flink介绍

Apache Flink是一个框架和分布式处理引擎，用于对无界和有界数据流进行有状态计算。
以内存执行速度和任意规模来执行计算。

中文文档：https://nightlies.apache.org/flink/flink-docs-stable/zh/

Flink有四个基石：Checkpoint机制（保障数据一致性）、State（提供API）、Time（数据容错）、Window（操作窗口）

1.1 Flink简介

1.1.1 Checkpoint

Checkpoint机制 为Flink实现了一个分布式的一致性的快照，从而提供了一致性的语义；

1.1.2 State

虽然有了一致性的语义之后，Flink为了让用户在编程时更加轻松、更容易地去管理状态，提供了一套非常简单明了的StateApi，包括里面的有ValueState、ListState、MapState，近期还添加了BroadcastState，使用State API能够自动先用这种一致性的语义。

1.1.3 Time

Flink还实现了Watemark的机制，能够支持基于事件的时间的处理，或者说基于系统时间的处理，能够容忍数据的延时、容忍数据的迟到、容忍乱序的数据。

1.1.4 Window

Flink提供了开箱即用的各种窗口，比如滑动窗口、滚动窗口、会话窗口以及非常灵活的自定义窗口。

1.2 Flink流处理特性

• 支持高吞吐、低延迟、高性能的流处理
• 支持带有事件时间的窗口（Window）操作
• 支持有状态计算的 Exactly-once 语义
• 支持高度灵活的窗口（Window）操作，支持基于 time、count、session，以及 data-driven 的窗口操作
• 支持具有 Backpressure 功能的持续流模型
• 支持基于轻量级分布式快照（Snapshot）实现的容错
• 一个运行时同时支持 Batch on Streaming 处理和 Streaming 处理
• Flink 在 JVM 内部实现了自己的内存管理
• 支持迭代计算
• 支持程序自动优化：避免特定情况下 Shuffle、排序等昂贵操作，中间结果有必要进行缓存

1.3 Flink的批处理和流处理

批处理：有界、持久、大量，批处理适合需要访问全套记录才能完成的计算工作，一般用于离线统计；
流处理：无界、实时，流处理无需针对整个数据集执行操作，而是对通过系统传输的每个数据项进行操作，一般用于实时统计；

Flink可以同时处理批处理和流处理。Flink通过将批处理（即处理有限的静态数据）视作一种特殊的流处理；

1.4 Flink Runtime执行引擎

Flink的核心计算架构是下图中的Flink Runtime执行引擎，他是一个分布式系统，能够接受数据流程序 并在一台或多台机器上以容错方式执行。

上图为Flink技术栈的核心组成部分，值得一提的是，Flink分别提供：
①面向流处理的接口（DataStream API）
②面向批处理的接口（DataSet API）
因此，Flink既可以完成流处理，也可以完成批处理。

Flink还支持的拓展库涉及：
①机器学习（FlinkML）
②复杂事件处理（CEP）
③图计算（Gelly）
④分别针对流处理和批处理的Table API

一一一一一
DataStream API可以流程地分析无限数据流，并且可以用Java或者Scala等来实现。开发者需要基于一个叫DataStream的数据结构来开发，这个数据结构用于表示永不停止的分布式数据流。

Flink的分布式特点体现在他能够在成百上千台机器上运行，它将大型的计算任务分成许多小的部分，每个机器执行一部分。
Flink能够自动地确保发生机器故障或者其他错误时计算能够持续进行，或者在修复bug或进行版本升级后有计划地再次执行一次。这种能力使得开发人员不需要担心运行失败。

Flink本质上使用容错性数据流，这使得开发人员可以分析持续生成切用不结束的数据（即流处理）；

二、Flink运行架构

2.1 Flink程序结构

Flink程序的基本构建块是 ：流和转换；（注意:DataSet API中使用的DataSet也是内部流）

从概念上讲，
流：可能永无止境的数据记录流；
转换：是将一个或多个流作为一个或多个流的操作。（即，输入并产生一个或多个输出流）

2.1.1 Source数据源

Flink在流处理和批处理上的source大概有四类：
① 基于本地集合的source；
② 基于文件的source；
③ 基于网络套接字的source；
④ 自定义的source；

自定义的source常见的有Apache kafka、RabbitMQ等，当然你也可以自定义自己的source；

2.1.2 Transformation

数据转换的各种操作，有Map/FlatMap/Filter/KetBy/Reduce/Fold/Aggregations/Window/WindowAll./Window join/Split/Select等，操作很多，可以将数据转换计算成你想要的数据。

2.1.3Sink

Sink接收器，Flink将转换计算后的数据发送的地点，可能需要存储下来。
Flink常见的Sink大概有如下几类：写入文件、打印出来、写入socket、自定义的sink。

自定义的sink常见的有Apache Kafka、RabbitMQ、MySQL、ElasticSearch、Apache Cassandra、Hadoop FileSystem等，同理也可以定义自己的sink。

2.2 Flink并行数据流

Flink程序本质上是并行的和分布式的；

Flink程序在启动、执行、结束的各个流程：
① 执行的时候，会被映射成一个Streaming Dataflow，一个Streaming Dataflow 是由一组Stream和Transformation Operator组成的。
② 启动时从一个或多个Source Operator开始
③ 结束于一个或多个Sink Operator。

Flink程序执行流程：执行的时候，会被映射成一个Streaming Dataflow，一个Streaming Dataflow 是由一组Stream和Transformation Operator组成的；
一个流Stream 包含一个或多个流分区，而每一个operator包含一个或多个operator子任务，也就是 operator subtask【操作子任务】。
operator subtask【操作子任务】间彼此独立，在不同的线程中执行，甚至是在不同的机器或不同的容器上。
operator subtask【操作子任务】的数量是这一特定operator的并行度。
相同程序中operator有不同级别的并行度。

一个Stream可以被分成多个Stream的分区，也就是Stream Partition【分区】。
一个Operator也可以被分为多个Operator Subtask【子任务】，如下图；

分析图片：

• Source被分为Source1 和 Source2，它们是Source的Operator Subtask
• 【也就是说Source1和Source2都是Source的 Operator Subtask子任务】。
• 每一个Operator Subtask都是在不同的线程当中独立执行的；
• 一个Operator的并行度，就等于Operator Subtask的个数；
• 下图Source的并行度为2，而一个Stream的并行度就等于它生成的Operator的并行度；
• 数据在两个 operator 之间传递的时候有两种模式： ①One to One 模式：两个operator用此模式传递的时候，会保持数据的分区数和数据的排序； 如下图中的Source1到IMap1，它就保留的Source的分区特性，以及分区元素处理的有序性。 ②Redistributing重新分配模式：这种模式会改变数据的分区数； 每一个Operator subtask【如下图的Source1 和 Source2】会根据选择Transformation把数据 发送到不同目标subtasks， 比如keyBy()会通过hashcode重新分区，broadcase()和rebalance()方法会随机重新分区；

2.3 Task和Operator chain

Flink的所有操作都被称之为Operator，客户端在提交任务的时候会对Operator进行优化操作，能进行合并的Operator会被合并为一个Operator，合并后的Operator称为Operator chain，实际上就是一个执行链，每个执行链会在TaskManager上一个独立的线程中执行。

2.4 任务调度与执行

• 当Flink执行executor会自动根据程序代码生成DAG数据流图；
• ActorSystem创建Actor将数据流图发送给JobManager的Actor；
• Jobmanager会不断接受TaskManager的心跳消息，从而可以获取到有效的TaskManager；
• JobManager通过调度器在TaskManager中调度执行Task（在Flink中，最小的调度单元就是task，对应就是一个线程）；
• 在程序运行过程中，task和task之间是可以进行数据传输的；

Job Client【就是上图的Flink程序】：

• 主要职责是提交任务，提交后可以结束进程，也可以等待结果返回；
• Job Client 不是Flink程序执行的内部部分，但它是任务执行的起点；
• Job Client负责接收用户的程序代码，然后创建数据流，将数据流提交给Job Manager以便进一步执行。执行完成后，Job Client将结果返回给用户；

JobManager：

• 主要职责是调度工作并协调任务做检查点；
• 集群中至少要有一个master，master负责调度task，协调checkpoints和容错；
• 高可用设置的话可以有多个master，但是要保证一个是leader，其他是standby；
• JobManager包含Actor System、Scheduler调度器、CheckPoint协调器 三个重要的组件；
• JobManager从客户端【上图的Flink程序】接收到任务后，首先生成优化过的执行计划，再调度到TaskManager中执行；

TaskManager

• 主要职责是从JobManager处接收任务，并部署和启动任务，接收上游的数据并处理；
• TaskManager是在JVM中的一个或多个线程中执行任务的工作节点；
• TaskManager在创建之初就设置好了Slot【槽】，每个Slot可以执行一个任务；

2.4.1任务槽和槽共享

每个TaskManager是一个JVM的进程，可以在不同的线程中执行一个或多个子任务。为了控制一个worker能接收多少个task。worker通过task slot【任务槽】来进行控制（一个worker至少有一个task slot）

任务槽：

• 每个task slot表示TaskManager拥有资源的一个固定大小的子集；
• flink将进程的内存进行了划分到多个slot中；
• 上图中有两个taskManager，每个TaskManager有三个slot，每个slot占1/3的内存；
• 内存被划分到不同的slot之后可以获得如下好处： ①TaskManager最多能同时并发执行的任务是可以控制的，那就是3个，因为不能超过slot的数量； ②slot有独占的内存空间，这样在一个TaskManager中可以运行多个不同的作业，作业之间不受影响；

槽共享：

• 只需计算Job中最高并行度（parallelism）的task slot，只要这个满足，其他的job也都能满足；
• 资源分配更加公平，如果有比较空闲的slot可以将更多的任务分配给它。图中若没有任务槽共享，负载不高的Source/Map 等subtask将会占据许多资源，而负载较高的窗口subtask则会缺乏资源；
• 有了任务槽任务，可以将基本并行度（base parallelism）从2提升到6，提高了分槽资源的利用率。同时还可以保障TaskManager给subtask的分配的slot方案更加公平；

三、Flink快速上手

3.1 准备

需求：统计一段文字中，每个单词出现的频次；
版本：基于1.17.0版本；
数据准备：在工程根目录下创建一个input文件夹，并且在下面创建文本文件words.txt，随便如下数据
批处理思路：先逐行读入文件数据，然后将每一行文字拆分成单词；接着按照单词分组，统计每组数据的个数，就是对应单词的频次；

3.2 导入依赖

<!-- fink 相关依赖 --><dependency><groupId>org.apache.flink</groupId><artifactId>flink-clients</artifactId><version>1.17.0</version></dependency><dependency><groupId>org.apache.flink</groupId><artifactId>flink-streaming-java</artifactId><version>1.17.0</version></dependency>

3.3 数据准备

查找到数据源：

思路：流处理，所以是一行一行的读取，然后按照空格切分，然后再分组统计；

3.4、批处理代码 ：实现wordcount 案例

DataSet API 批处理 （过时了）

packagecom.flink17.demo;importorg.apache.flink.api.common.functions.FlatMapFunction;importorg.apache.flink.api.java.ExecutionEnvironment;importorg.apache.flink.api.java.operators.AggregateOperator;importorg.apache.flink.api.java.operators.DataSource;importorg.apache.flink.api.java.operators.FlatMapOperator;importorg.apache.flink.api.java.operators.UnsortedGrouping;importorg.apache.flink.api.java.tuple.Tuple2;importorg.apache.flink.util.Collector;/**
 * DataSet API 实现wordcount
 *
 * @author lc
 * @version 1.0
 * @date 2024/10/8 0008 16:27
 */publicclassWordCountStreamDemoMain{publicstaticvoidmain(String[] args)throwsException{// 1.创建执行环境ExecutionEnvironment env =ExecutionEnvironment.getExecutionEnvironment();// 2.读取数据（这里是文本数据）DataSource<String> lineDS = env.readTextFile("input/words.txt");// 3.切分、转换（word,1）的格式（第一个参数 word表示的是单词，第二个参数 1是指出现的次数）// flatMap的方法中的FlatMapFunction 的第一个参数是输入，第二个参数是输出FlatMapOperator<String,Tuple2<String,Integer>> wordAndOne
                = lineDS.flatMap(newFlatMapFunction<String,Tuple2<String,Integer>>(){// 重写 flatMap ，第一个参数是需要操作的数据，第二个参数Collector是采集器@OverridepublicvoidflatMap(String s,Collector<Tuple2<String,Integer>> collector)throwsException{//3.1 按照空格切割String[] words = s.split(" ");//3.2 将单词转换成二元组（word，1）这样的格式for(String word : words){Tuple2<String,Integer> wordTuple2 =Tuple2.of(word,1);// 3.3 使用Collector向下游发送数据
                    collector.collect(wordTuple2);}}});// 4.按照word分组UnsortedGrouping<Tuple2<String,Integer>> wordAndOneGroupBy = wordAndOne.groupBy(0);// 5.各分组内聚合AggregateOperator<Tuple2<String,Integer>> sum =
                wordAndOneGroupBy.sum(1);// 这里的1 是位置，表示第二个元素// 6.输出
        sum.print();}}

执行结果

DataSet写法实现批处理是过时的，推荐使用DataStream来写；

3.5、流处理代码（有界流-有开始有结束）：：实现wordcount 案例

packagecom.flink17.demo;importorg.apache.flink.api.common.functions.FlatMapFunction;importorg.apache.flink.api.java.functions.KeySelector;importorg.apache.flink.api.java.tuple.Tuple2;importorg.apache.flink.streaming.api.datastream.DataStreamSource;importorg.apache.flink.streaming.api.datastream.KeyedStream;importorg.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;importorg.apache.flink.streaming.api.environment.StreamExecutionEnvironment;importorg.apache.flink.util.Collector;/**
 * DataStream流处理 实现wordcount：（读文件，有界流）
 *
 * @author lc
 * @version 1.0
 * @date 2024/10/8 0008 16:53
 */publicclassWordCountStreamDemoMain{publicstaticvoidmain(String[] args)throwsException{// 1.创建执行环境StreamExecutionEnvironment evn =StreamExecutionEnvironment.getExecutionEnvironment();// 2.读取数据DataStreamSource<String> lineDs = evn.readTextFile("input/words.txt");// 3.处理数据:切分、转换、分组、聚合SingleOutputStreamOperator<Tuple2<String,Integer>> wordAndOne = lineDs.flatMap(newFlatMapFunction<String,Tuple2<String,Integer>>(){@OverridepublicvoidflatMap(String value,Collector<Tuple2<String,Integer>> out)throwsException{String[] words = value.split(" ");// 空格切分for(String word : words){// 转换成二元组Tuple2<String,Integer> wordsAndOne =Tuple2.of(word,1);// 使用Collector向下游发送数据
                    out.collect(wordsAndOne);}}});// 分组KeyedStream<Tuple2<String,Integer>,String> ks = wordAndOne.keyBy(// 第一个参数书数据的类型，第二个参数是key的类型newKeySelector<Tuple2<String,Integer>,String>(){@OverridepublicStringgetKey(Tuple2<String,Integer> value)throwsException{return value.f0;}});SingleOutputStreamOperator<Tuple2<String,Integer>> sum = ks.sum(1);// 这里的1 是位置，表示第二个元素// 4.输出数据
        sum.print();// 5.执行
        evn.execute();}}

执行结果

流处理：拿一条处理一条；所以是一行一行的读取；且有状态，比如hello最后为（hello，3）3是有状态的计算；
前面的编号就是并行度编号，也就是线程数编号；

批处理：一口气处理一批（这里就是整个文本）

3.6、流处理代码（无界流-有开始没有结束，更加常用）：：实现wordcount 案例

在实际的生产环境中，真正的数据流其实是无界的，有开始却没有结束，这就要求我们需要持续地处理捕获的数据；
模拟场景：监听socket端口，然后向该端口不断的发送数据

将StreamWorldCount代码中读取文件数据的readTextFile方法替换成socket文本流的方法socketTextStram；

packagecom.flink17.demo;importorg.apache.flink.api.common.typeinfo.Types;importorg.apache.flink.api.java.tuple.Tuple2;importorg.apache.flink.streaming.api.datastream.DataStreamSource;importorg.apache.flink.streaming.api.environment.StreamExecutionEnvironment;importorg.apache.flink.util.Collector;/**
 * @author lc
 * @version 1.0
 * @date 2024/10/8 0008 17:17
 */publicclassWordCountStreamUnboundDemoMain{publicstaticvoidmain(String[] args)throwsException{//1.创建执行环境StreamExecutionEnvironment env =StreamExecutionEnvironment.getExecutionEnvironment();//2.读取数据：socket,第一个参数书服务器名称，第二个参数是端口号DataStreamSource<String> socketDS = env.socketTextStream("hadoop102",7777);//3.处理数据：切分，转换，分组，聚合，输出
        socketDS
                .flatMap((String value,Collector<Tuple2<String,Integer>> out)->{String[] words = value.split(" ");// 空格切分for(String word : words){// 转换成二元组Tuple2<String,Integer> wordsAndOne =Tuple2.of(word,1);// 使用Collector向下游发送数据
                                out.collect(wordsAndOne);}}).returns(Types.TUPLE(Types.STRING,Types.INT))//需要执行返回类型，（word，1）.keyBy(value -> value.f0).sum(1).print();// 输出//4.执行
        env.execute();}}

不指定返回类型的化，执行会报错；因为Flink具有一个类型提取系统，可以分析函数的输入和返回类型，自动获取类型信息，从而获得对应序列化器和反序列化器，但是，由于java中泛型查出的存在所以报错了；

执行，就会一直监听那个服务器，有输入才会有响应的输出；

有界流是有结束的，无界留是没有结束的

四、Flink部署及启动

4.1、Flink的执行逻辑

Flink提交作业和执行任务，需要几个关键组件：
1.客户端（client）：代码由客户端获取并做转换，之后提交给JobManager
2.JobManager 就是 Flink集群里面的“管事人”，对作业进行中央调度管理；而他获取到要执行的作业后，会进一步处理转换，然后分布任务给众多的TaskManager；
3.TaskManager，就是真正“干活的人”，数据的处理操作 都是他们来做的；

4.2、Flink的安装模式

Flink支持多种安装模式：
1.local-本地安装：单机模式，一般不使用；
2.standalone-独立模式：Flink自带集群，开发测试环境使用；
3.yarn：计算资源统一由Hadoop YARN管理，生产环境使用；

4.2.1、准备工作：

Flink 是一个分布式的流处理框架，所以实际应用一般都需要搭建集群环境。我们在进行
Flink 安装部署的学习时，需要准备 3 台 Linux 机器。
具体要求如下：
    系统环境为 CentOS 7.5 版本（也可以是Ubuntu）
    安装 Java 8。

4.2.1.1 Flink和Java下载安装

安装Flink：
官网：https://flink.apache.org/downloads/
清华镜像：https://mirrors.tuna.tsinghua.edu.cn/apache/flink/

1.可以从官网下载Flink的安装包，如下1.20.0版本的；
2.下载完后可以移动到自己想存储的位置

3.解压安装包


4.启动flink

进入到解压目录下，执行以下脚本：

发现报错，是因为没有安装jdk，这里我们安装jdk8；

5.安装jdk

网址：https://www.oracle.com/java/technologies/downloads/#java8
jdk安装：
ubuntu中自带了jdk，先将其卸载：sudo apt-get remove openjdk
sudo apt-get autoremove

上传安装包到自指定路径：/usr/local/jdk

解压安装包：tar -zxvf jdk-8u291-linux-x64.tar -C /usr/local/myapp/jdk

配置环境变量：sudo vim /etc/profile
ESC然后再 : + shfit 输入wq，保存退出
在文末增加配置（路径根据实际情况调整）：
export JAVA_HOME=/usr/local/myapp/jdk/jdk1.8.0_291
export JRE_HOME=/usr/local/myapp/jdk/jdk1.8.0_291/jre
export PATH=

     P 
    
   
     A 
    
   
     T 
    
   
     H 
    
   
     : 
    
   
  
    PATH: 
   
  
PATH:JAVA_HOME/bin: 
 
  
   
   
     J 
    
   
     R 
    
    
    
      E 
     
    
      H 
     
    
   
     O 
    
   
     M 
    
   
     E 
    
   
     / 
    
   
     b 
    
   
     i 
    
   
     n 
    
   
     e 
    
   
     x 
    
   
     p 
    
   
     o 
    
   
     r 
    
   
     t 
    
   
     C 
    
   
     L 
    
   
     A 
    
   
     S 
    
   
     S 
    
   
     P 
    
   
     A 
    
   
     T 
    
   
     H 
    
   
     = 
    
   
  
    JRE_HOME/bin export CLASSPATH= 
   
  
JREH​OME/binexportCLASSPATH=CLASSPATH: 
 
  
   
   
     J 
    
   
     A 
    
   
     V 
    
    
    
      A 
     
    
      H 
     
    
   
     O 
    
   
     M 
    
   
     E 
    
   
     / 
    
   
     l 
    
   
     i 
    
   
     b 
    
   
     : 
    
   
  
    JAVA_HOME/lib: 
   
  
JAVAH​OME/lib:JRE_HOME/lib

测试jdk：java -version 或者 javac -version
root@vm1:/usr/local/myapp/jdk# java -version
java version “1.8.0_291”
Java™ SE Runtime Environment (build 1.8.0_291-b10)
Java HotSpot™ 64-Bit Server VM (build 25.291-b10, mixed mode)
root@vm1:/usr/local/myapp/jdk# javac -version
javac 1.8.0_291

6. Flink安装包然后解压到指定目录,注意修改所属用户和用户组

#.改名
mv flink-1.7.2 flink
#.赋予权限
chown -R root:root flink

7.再次启动

关闭防火墙：sudo ufw disable

启动成功，通过jps查看服务信息：
StandaloneSessionClusterEntrypoint为Flink主进程，即JobManager；TaskManagerRunner为Flink从进程，即TaskManager

8.页面查看
单机启动，自己既是jobmanager，也是taskmanager

在浏览器中访问服务器8081端口即可查看Flink的WebUI，
比如http://localhost:8081/，从WebUI中可以看出，当前本地模式的Task Slot数量和TaskManager数量。访问结果如下图所示：

9.停止flink
./bin/stop-cluster.sh
刷新页面会报错

10.可以通过观察logs目录下的日志来检测系统是否正在运行了
tail log/flink–jobmanager-.log

4.2.2、单节点部署

4.2.3、standalone安装模式

我这边是给Flink003为Master机器（JobManager），其他的为Slave机器（TaskManager）

4.2.3.1 ssh配置

集群的服务器之间配置好ssh免密登录，避免后续搭建出现麻烦，这一步一定要做。简单步骤如下：

1. 先进入root用户权限 sudo -i，且修改root用户的密码 输入passwd

2. 查看服务器是否安装有ssh服务

1. 生成密钥和公钥；和在master机器执行ssh-keygen -t rsa
2. 把公钥复制到需要免密登录的从服务器上；在master机器执行命令，将密钥拷贝到其余服务器ssh-copy-id -i /root/.ssh/id_rsa.pub 目标服务器IP （报错的化看下图的原因解释）

我们执行ssh-copy-id 报错，说连接被拒，可以进行排错

确认目标主机的SSH服务是否运行：sudo systemctl status ssh，如果未运行执行：sudo systemctl start ssh；
![
那就是ssh服务没有被安装，需要安装，ubuntu上检查安装状态输入 sudo apt-listfiles | grep openssh-server，
在使用sudo apt-get install openssh-server安装；

检查SSH配置，确保SSH服务的配置文件 /etc/ssh/sshd_config正确，特别是监听的端口设置是Port 22，如果使用了自定义端口，确保该端口正确配置并且未被防火墙阻挡。

可以之后再试ssh-copy-id ；

1. 在继续ssh-copy-id ，输入密码；
2. 验证免密登录。运行以下命令连接到远程服务器：ssh username@remote_server_ip。如果成功连接而无需输入密码，则表示设置成功

要退出ssh连接输入exit就可以了；

4.2.3.2 修改Flink配置，实现基础的集群搭建

1. 修改flink配置文件： 进入Flink的配置文件，conf目录下的flink-conf.yaml 或者cong.yaml: vim conf.yaml 将文件中的jobmanager.rpc.address属性进行修改为JobManager机器也就是主机的ip地址，

jobmanager.rpc.address: 主机ip地址
jobmanager.bind-host: 0.0.0.0

2.修改workers文件
vim conf/workers
workers文件必须包含所有需要启动的TaskManager节点的主机名，且每个主机名占一行。在JobManager服务器，执行以下操作


3. 复制Flink安装文件到其他服务器

其他从服务器不需要下载flink的安装包，必须从主scp到从，因为有路径要求，才能关联起来；

scp -r /usr/local/flink/ root@从服务器1的ip地址：/usr/local/flink/
scp -r /usr/local/flink/ root@从服务器2的ip地址：/usr/local/flink/

4.集群启动
这里只需要在主服务器上，也就是JobManager的服务器上启动Flink，从服务也会跟着启动；
在JobManager节点上进入Flink安装目录，执行以下命令启动Flink集群：

启动完毕后，在集群各服务器上通过jsp命令查看Java进程。若各节点存在以下进程，则说明集群启动成功：
JobManager节点：StandaloneSessionClusterEntrypoint
TaskManager1节点：TaskManagerRunner
TaskManager2节点：TaskManagerRunner

我们输入从机的密码，可以看到提醒我们启动了从机的flink（这里我只弄了一台机器从机），下图是主机的命令窗口，输入jps可以看到作为JobManager的StanaloneSessionClusxxxx启动了

从机命令窗口输入jps可以看到TaskManager启动成功；

尝试提交一个简单任务，如果任务正常执行完毕，则集群一切正常。提交Flink自带的简单任务如下：
./bin/flink run examples/streaming/WordCount.jar

5.查看webUI
通过JobManager节点访问WebUI（http://localhost:8081/），可以看到此时是1个JobManager，1个TaskManager，也能以上执行完毕的任务，如下图：

4.2.4 进阶版集群-Flink Standalone HA搭建

在Flink Standalone模式下，实现HA的方式可以利用ZooKeeper在所有正在运行的JobManager实例之间进行分布式协调，实现多个JobManager无缝切换。Flink Standalone模式的HA架构如图：

简单来说就是 使用 ZooKeeper 来实现JobManager无缝切换，从而达到分布式协调；

注意：Flink内置了Zookeeper服务和相关脚本文件，如果你的集群中没有安装Zookeeper，则可以通过修改zoo.cfg文件配置Flink内置的Zookeeper。生产环境建议使用独立的外部zookeeper；

HA的核心就是：可以在集群中启动多个JobManager，并使它们都向ZooKeeper进行注册，ZooKeeper利用自身的选举机制保证同一时间只有一个JobManager是活动状态（Active）的，其他的都是备用状态（Standby）。当活动状态的JobManager出现故障时，ZooKeeper会从其他备用状态的JobManager选出一个成为活动JobManager。流程见下图：

此外，活动状态的JobManager在工作时会将其元数据（JobGraph、应用程序JAR文件等）写入一个远程持久化存储系统（例如HDFS）中，还会将元数据存储的位置和路径信息写入ZooKeeper存储，以便能够进行故障恢复，如图下图所示：

操作起来！！！！！
首先，我们还是在上面的操作的三个服务器进行操作；
目前是一个jobmanager（Flink003机器），两个taskmanager（Flink001和Flink002机器）；
现在我们需要变成两个jobmanager，一个taskmanager；

1. 修改masters文件 Flink的masters文件用于配置所有需要启动的JobManager节点以及每个JobManager的WebUI绑定的端口。

进入Flink安装目录，修改conf/masters文件，修改内容如下：

Flink003机器的ip地址:8081
Flink002机器的ip地址:8082

上述配置表示在集群 Flink003机器 和 Flink002机器节点上启动JobManager，并且每个JobManager的WebUI访问端口分别为8081和8082。

1. 修改flink-conf.yaml文件设置高可用模式 进入Flink003机器 节点的Flink安装主目录，修改conf/flink-conf.yaml文件，添加以下内容：

# 将高可用模式设置为ZooKeeper，默认集群不会开启高可用状态
high-availability: zookeeper
# ZooKeeper集群主机名（或IP）与端口列表，多个以逗号分隔
high-availability.zookeeper.quorum: centos01:2181,centos02:2181,centos03:2181
# 用于持久化JobManager元数据（JobGraph、应用程序JAR文件等）的HDFS地址，以便进行故障恢复，ZooKeeper上存储的只是元数据所在的位置路径信息
high-availability.storageDir: /data/software/flink-15.4/ha
# 获取storageDir也可用hdfs，如果使用hdfs的话，则需要单独安装hdfs，本文暂不使用
#high-availability.storageDir: hdfs://centos01:9000/flink/recovery
 

改动如图下

1. 修改zoo.cfg文件 Flink内置了ZooKeeper服务和相关脚本文件，如果你的集群中没有安装ZooKeeper，则可以通过修改zoo.cfg文件配置Flink内置的ZooKeeper。生产环境建议使用独立的外部ZooKeeper。

进入centos01节点的Flink安装主目录，修改conf/zoo.cfg文件，添加以下内容，配置ZooKeeper启动节点与选举相关端口：

server.1=centos01:2888:3888
server.2=centos02:2888:3888
server.3=centos03:2888:388

上述配置表示在centos01、centos02和centos03节点上启动ZooKeeper服务，其中1、2、3表示每个ZooKeeper服务器的唯一ID。

1. 复制Flink安装文件到其他节点 继续采用scp命令，复制centos01的文件到其他节点，scp命令会把相同文件覆盖。

scp -r /usr/local/flink/ root@从服务器1的ip地址：/usr/local/flink/
scp -r /usr/local/flink/ root@从服务器2的ip地址：/usr/local/flink/

1. 启动ZooKeeper集群 如果使用Flink内置的ZooKeeper，在Flink003机器节点执行以下命令，即可启动整个ZooKeeper集群： …/bin/start-zookeeper-quorum.sh

启动成功后，在每个Flink节点上都会产生一个名为FlinkZooKeeperQuorumPeer的进程，该进程是ZooKeeper服务的守护进程。使用jps可以查看到如下进程：



注意：这里本级需要使用localhost
6. Flink003机器节点上执行以下命令，启动Flink Standalone HA集群：

要先启动zookeeper再启动flink

bin/start-cluster.sh

Flink003主机 jps：

Flink002和Flink001机器 jps：

1. 访问webUI 以前只有Flink003可以访问webui页面也就是Flink的Dashboard，现在是Flink003和Flink002都可以访问

假如方向访问不了，说明配置不对，重新查看自己的配置，要注意改完Flink003的后要从新scp复制到从服务器上；

8.测试
在提交一个测试，如果能正常执行，说明整个集群正常。

./bin/flink run examples/streaming/WordCount.jar

9.停止集群

若要停止Flink Standalone HA集群，在jobmanager节点上首先执行以下命令停止整个Flink集群，我这里是flink003：

./stop-zookeeper-quorum.sh

4.2.5、Yarn安装模式

完成上述的Flink和jdk的安装后；

进入Flink的conf目录，依据Flink的版本进行修改flink-conf.yaml或者conf.yaml文件

bind-host 就是0.0.0.0
rpc.address指的是从机的ip地址