Flink运行模式

    在idea中运行Flink程序的方式就是开发模式。

    Flink中的Local-cluster(本地集群)模式,单节点运行，主要用于测试, 学习。

        独立集群模式，由Flink自身提供判差计算资源。

把Flink应用提交给Yarn的ResourceManager

Flink会根据运行在JobManger上的job的需要的slot的数量动态的分配TaskManager资源

Yarn又分3种模式

Session-Cluster模式需要先启动Flink集群，向Yarn申请资源。以后提交任务都向这里提交。

这个Flink集群会常驻在yarn集群卖磨中，除非手工停止。

在向Flink集群提交Job的时候, 如果资源被用完了,则新的Job不能正常提交.

缺点: 如果提交的作业中有长时间执行的大作业, 占用了该Flink集群的所有资源, 则后续无法提交新的job.

所以, Session-Cluster适合那些需要频繁提交的多个小Job, 并且执行时间都不长的Job.

一个Job会对应一个Flink集群，每提交一个作业会根据自身的情况，都会单独向yarn申请资源，直到作业执行完成，一个作业的失败与否并不会影响下一个作业的正常提交和运行。独享Dispatcher和ResourceManager，按需接受资源申请；适合规模大长时间运行的作业。

每次提交都会创建一个新的flink集群，任务之间互相独立，互不影响，方便管理。任务执行完成之后创建的集群也会消失。

Per-job模式执行结果，一个job对应一个Application

Application Mode会在Yarn上启动集群, 应用jar包的main函数(用户类的main函数)将会在JobManager上执行. 只要应用程序执行结束, Flink集群会马上被关闭. 也可以手动停止集群.

与Per-Job-Cluster的区别：就是Application Mode下, 用户的main函数式在集群中执行的，并且当一个application中有多个job的话，per-job模式则是一个job对应一个yarn中的application，而Application Mode则这个application中对应多个job。

Application Mode模式执行结果，多个job对应一个Application

官方建议:

出于生产的需求, 我们建议使用Per-job or Application Mode,因为他们给应用提供了更中冲斗好的隔离!

https://ci.apache.org/projects/flink/flink-docs-release-1.12/deployment/

0．Flink任务提交后，Client向HDFS上传Flink的Jar包和配置

1．向Yarn ResourceManager提交任务，ResourceManager分配Container资源

2．通知对应的NodeManager启动ApplicationMaster，ApplicationMaster启动后加载Flink的Jar包和配置构建环境，然后启动JobManager（Dispatcher）

2.1．Dispatcher启动JobMaster

3．JobMaster向ResourceManager（Flink）申请资源

4．ResourceManager（Flink）向ResourceManager（Yarn）申请资源启动TaskManager

5．ResourceManager分配Container资源后，由ApplicationMaster通知资源所在节点的NodeManager启动TaskManager

6．TaskManager注册Slot

7．发出提供Slot命令

8．TaskManager向JobMaster提供Slot

9．JobMaster提交要在Slot中执行的任务

在Flink进行数据处理的时候，有两个最重要的两个组件，分别是：作业管理器（JobManager）和任务管理器（TaskManager）。对于一个提交执行的作业，JM是真正的管理者，负责管理和调度工作。如果集群没有配置高可用的话，只能有一个JM。TM是负责工作的工作者，负责执行任务和处理数据，所以可以有很多个。

整个JM由三个部分组成，分别是JobMaster、ResourceManager、Dispatcher。

JMaster是JManager中最核心的组件，负责处理单独的作业。所以每一个Job都会有一个对应JobMaster与之对应，多个job可以同时运行在一个FLink集群中，也就证明一个Flink集群中可以有多个JobMaster同时存在。在作业被提交的时候，JMaster会率先接受到要执行的应用。这些应用包括：Jar包、数据流图（dataflow graph）、作业图（JobGraph）。然后将接收到的作业图转换成为一个物理层面的数据流图，也就是执行图，这个执行图包含了所有要并发执行的任务。随后JMaster会根据这个执行图去到RM中申请必要的资源。一旦获得到了足够的资源举岩，就会将执行图发送到要运行执行图的TaskMaster上面。并且在程序运行的过程中，JobMaster会负责协调所有的需要中央协调的工作，如保存点机制。

RM在Flink的集群中只有一个，它主要负责的是资源的分配与管理，也就是对TaskManager上的solt的分配工作。并且在面对不同的环境的时候，RM也有不同的体现。如果是standalone，因为TM是独立运行的，所有RM只能分发可用的TM的任务槽，没办法单独启动TM。但是如果是yarn的工作模式下，RM能够将有空闲的TM发送给JMaster，如果资源不够使用的话，就会向yarn发起请求提供启动TM的容器。并且如果有的TM处于空闲状态，RM还负责停止它们的工作。

它主要负责提供Rest风格的接口来提交应用。并且还负责为每一个新提交的作业启运老动一个新的JMaster组件，并且还会启动一个WEB UI，方便用来展示和键控作业执行的信息。

TM负责FLink集群中的具体工作部分，数据流的计算就是由这个组件来完成的，每一个TM都会有一定数量的任务槽，这个任务槽是一个TM上最小的资源封装单位，solt的数量决定了TM能够并行处理的任务的数量。在TM启动之后，会向RM注册自己的solt，当收到RM对其发送的指令之后，就会根据要求发送能够提供计算的solt给JMaster调用，这样数正悄御据计算就能够实现了，并且在进行计算的过程中，TaskManager可以缓冲数据，还能够与其他的TM交互完成数据的交换。

因为我本人所接触的Flink的部署模式都是基于资源管理平台yarn来实现工作的，采用的作业提交方式也是通过per-job提交方式进行提交的，所以在本次讲述的过程中，也是以这个内容为蓝本展开讲解。

yarn-per-job作业提交流程：

在单作业模式下，Flink集群不会预先启动，而是在进行作业提交的时候，才会启动新的JobManager。

1.客户端向yarn提交作业，并且需要将Flink的Jar包和配置文件信息上传到HDFS，以便后续启动FLink相关组件的容器。

2.YARN资源管理器分配Container资源，启动Application Master，这个APP里面包含了Flink的JobManager，并且要将提交上来的作业交给JMaster。

3.JMaster向flink的rm请求资源。

4.flink的rm向yarn请求container资源。

5.yarn启动包含TM的container资源。

6.TM向JMaster注册自己拥有的solt数量。

7.flink的RM向TM申请solt。

8.TM连接到对应的JMaster，然后通过solt。

9.JMaster将要执行的任务分发给TM，执行。

注意:系统和运行脚本在启动时解析配置.对配置文件的更改需要重新启动Flink JobManager和TaskManagers

Flink on Yarn模式安装部署要做的其实不多，正常的步骤：

1、上传二进制包 ===》2、解压缩 ===》 3、更改文件名称 ===》 4、配置环境变量。Flink on yarn的job运行模式大致分为两类：

第一种巧知模式分为两步：yarn-session.sh(开辟资源)--->flink run（提交任务）

另外，jobmanager和taskmanager分别占有容器，示例：

./bin/yarn-session.sh -n 10 -tm 8192 -s 32

上面的例子将冲让会启动11个容器(即使仅请求10个容器)，因为有一个额外的容器来启动ApplicationMaster 和 job manager,一旦flink在你的yarn集群上部署，它将会显示job manager的连接详细信息。

第二种模式其实也孝判消分为两个部分，依然是开辟资源和提交任务，但是在Job模式下，这两步都合成一个命令了。

这里，我们直接执行命令

在job结束后就会关闭flink yarn-session的集群

sudo /usr/lib/flink/bin/flink run -m yarn-cluster -yn 1 -yjm 1024 -ytm 1024 -ys 1 -p 1 xz-flink-examples-1.0.jar

• "run" *** 作参数:

注意：client必须要设置YARN_CONF_DIR或者HADOOP_CONF_DIR环境变量，通过这个环境变量来读取YARN和HDFS的配置信息，否则启动会失败。

经试验发现，其实如果配置的有HADOOP_HOME环境变量的话也是可以的。HADOOP_HOME ，YARN_CONF_DIR，HADOOP_CONF_DIR 只要配置的有任何一个即可。

独立job模式客户端命令行参数参考： flink独立Job命令

Flink 的 YARN 客户端具有以下配置参数来控制容器故障时的行为方式。这些参数可以从 conf/flink-conf.yaml 中设置，或者在启动会话时使用-D参数设置

如：

参考： flink中文官网关于参数的解释

---