Flink部署模式及核心概念

一.部署模式

1.1会话模式（Session Mode）

需要先启动一个 Flink 集群，保持一个会话，所有提交的作业都会运行在此集群上，且启动时所需的资源以确定，无法更改，所以所有已提交的作业都会竞争集群中的资源。 

1.2单作业模式（Per-Job Mode） 

会为每一个作业单独启动一个集群，且作业完成或关闭后，相应的资源也会被释放。

1.3应用模式（Application Mode）

前两种模式下，应用代码都是在客户端上执行的，然后再由客户端交由 JobManager，客户端需要下载依赖，再将二进制数据发送给 JobManager ，这会加重客户端所在节点的资源压力，而且会占用大量的网络带宽。

所以我们就可以直接将应用提交到 JobManager ，而不通过客户端。引用模式与单作业模式类似，只不过将应用直接交给 JboManager 执行。

二.Flink运行时架构

2.1 作业管理器（JobManager）

JobManager是一个Flink集群中任务管理和调度的核心，是控制应用执行的主进程。且每个应用都应该被一个唯一的JobManager所控制执行。JobManger又包含3个不同的组件：

1、JobMaster

        是JobManager中最核心的组件，负责处理单独的作业（Job）。每个Job都有一个自己的JobMaster。

2.资源管理器（ResourceManager

        负责资源的分配和管理，在Flink 集群中只有一个。所谓“资源”，主要是指TaskManager的任务槽（task slots）。任务槽就是Flink集群中的资源调配单元，包含了机器用来执行计算的一组CPU和内存资源。每一个任务（Task）都需要分配到一个slot上执行。

3.分发器（Dispatcher）

        主要负责提供一个REST接口，用来提交应用，并且负责为每一个新提交的作业启动一个新的JobMaster 组件。

2.2 任务管理器（TaskManager）

askManager是Flink中的工作进程，数据流的具体计算就是它来做的。每一个TaskManager都包含了一定数量的任务槽（task slots）。Slot是资源调度的最小单位，slot的数量限制了TaskManager能够并行处理的任务数量

三.核心概念

3.1 并行度（Parallelism）

3.1.1 并行子任务和并行度

算子其实就是对流的操作，例如FlatMap、Sum等。

当需要处理的数据非常大时，我们可以将一个算子操作复制到多个节点执行，这样一个算子任务就被拆分成了多个“子任务（subtasks）”，再将它们分发到不同节点，就真正实现了并行计算。换句话来说就是一个算子操作不止一个人在干活，几个人干活就对应并行度。

 Flink定义一个流程序的并行度，可以认为就是其所有算子中最大的并行度。

3.1.2 并行度的设置

在 Flink 中，有不同的方式来设置并行度，且他们的优先级也有区别。

1.代码中设置

        可以直接在算子后调用 setParallelism() 方法来设置当前算子的并行度。

stream.map(word -> Tuple2.of(word, 1L)).setParallelism(2);

         也可以在创建执行环境时设定全局并行度。

env.setParallelism(2);

2.提交应用时设置

• 在使用 Flink run 命令在命令行提交任务时，可使用 -p 来指定当前应用程序执行的并行度。
• 在WebUI中提交任务时，手动指定并行度。
• 在 Flink的 flink-conf.yaml 配置文件下的参数 parallelism.default 可配置默认并行度。

3.2 算子链（Operator Chain）

3.2.2 算子间的数据传输

        一个数据流在算子之间传输数据的形式可以是一对一（one-to-one）的直通（forwarding）模式，也可以是打乱的重分区（redistributing）模式，具体是哪一种形式，取决于算子的种类。

• 一对一（One-to-one，forwarding）

        在这种模式下，数据流还是保持着原来的分区和顺序，例如Source算子读取数据后，直接交给Map算子，它们之间不需要重新分区，也不需要调整顺序，这种就被称为“一对一”。例如 map、filter、flatMap等算子都是这种one-to-one的对应关系。

重分区（Redistributing）

        在这种模式下，数据流的分区或顺序会发生变化，例如使用了 KeyBy算子。

3.2.3 合并算子链

在 Flink 中，并行度相同且为一对一的算子操作，可以直接合并为一个大的任务(Task)，这样原来的算子就成为了真正任务里的一部分，每个task会被一个线程执行。这样的技术被称为“算子链”（Operator Chain）。

例如有以下操作：

Source -> Map -> keyBy -> Sink

假设所有算子并行度为 2 ，由于 Source 和 Map 满足了算子链的要求（ 并行度相同且为一对一） ，则可以进行合并算子链。合并算子链后， Source 的一个子任务则会和 Map 的一个子任务合并成一个子任务， Source 和 Map 在未开启合并算子链前的子任务个数和为4，而开启合并算子链后的子任务个数和为2。

将算子链接成task是非常有效的优化：可以减少线程之间的切换和基于缓存区的数据交换，在减少时延的同时提升吞吐量。

在一些算子操作中计算量较重，或者需要精确排查错误，则可以考虑将某些算子的合并算子链关闭，不过 Flink 会默认为我们进行优化。

// 禁用算子链 map不会与前一个算子、后一个算子相连成一个算子链
.map(word -> Tuple2.of(word, 1L)).disableChaining();
 
// 从当前算子开始新链 与前一个算子断开，与后面的算子相连成一个算子链
.map(word -> Tuple2.of(word, 1L)).startNewChain()

3.3 任务槽（Task Slots）

3.3.1 任务槽（Task Slots）

Flink中每一个TaskManager都是一个JVM进程，它可以启动多个独立的线程，来并行执行多个子任务（subtask）。

TaskManager的计算资源是有限的，并行的任务越多，每个线程的资源就会越少。为了控制每个TaskManager的资源，则需要在TaskManager上对每个任务运行所占用的资源做出明确的划分，这就是所谓的任务槽（task slots）。

每个任务槽（task slot）其实表示了TaskManager拥有计算资源的一个固定大小的子集。这些资源就是用来独立执行一个子任务的。

假设一个TaskManager有三个slot，那么它会将管理的内存分为3份，每个slot独享一份。这样一来，我们在slot执行一个子任务时，那么这个子任务就会独享一块内存，而不需要去和其他的任务竞争内存资源了。

不同的算子必须在不同在 Slot 中。

3.3.2 任务槽数量的设置

在 Flink 的 flink-conf.yaml 配置文件下的参数 taskmanager.numberOfTaskSlots 可配置默认并行度（默认为1）。      

注意：slot目前仅仅用来隔离内存，不会涉及CPU的隔离。在具体应用时，可以将 slot 数量配置为机器的CPU核心数，尽量避免不同任务之间对CPU的竞争。这也是开发环境默认并行度设为机器CPU数量的原因。  

3.3.3 任务对任务槽的共享

在同一个作业中，不同任务节点（算子）的并行子任务可以被放到同一个 Slot 上执行。

在同一个 Slot 中运行的算子必须都是来自不同的算子，且同一个 Slot 中的算子都是同时运行的。

Flink默认是允许slot共享的，如果希望某个算子对应的任务完全独占一个slot，或者只有某一部分算子共享slot，也可以通过设置“slot共享组”手动指定： 

只有属于同一个slot共享组的子任务，才会开启slot共享；不同组之间的任务是完全隔离的，必须分配到不同的slot上。默认的共享组为“default”。在这种场景下，总共需要的slot数量，就是各个slot共享组最大并行度的总和。

3.3.4 任务槽和并行度的关系

Slot 是一种静态的概念，表示最大的并发上线。并行度是一种动态的概念，表示实际运行所占用的 Slot 个数。

假设有三个 TaskManager，且并行度为3，则表示集群最多能并行执行9个同一算子的子任务。

定义 Flink 程序由以下转换算子：

source→ flatmap→ reduce→ sink

当他们的并行度相同时，source 和 flatmap 则可以合并算子链，于是就有三个任务节点，且可以放到一个 Slot 中。

并行度为1：

并行度为9：

 并行度为9，但 sink 并行度为1：

当 Job并行度（算子最大并行度） 大于 Slot 数量时，Job将无法运行。 但是使用 Yarn 模式可以动态根据所需的 Solt 数量分配足够数量的 JobManager。

3.4 作业提交流程

3.4.1 Standalone会话模式作业提交流程

 逻辑流图/作业图/执行图/物理流图

代码生成的任务的过程：逻辑流图（StreamGraph）→ 作业图（JobGraph）→ 执行图（ExecutionGraph）→ 物理图（Physical Graph）。

3.4.2 Yarn应用模式作业提交流程