系统架构

Flink运行时架构Standalone会话模式为例

1）作业管理器（JobManager）

JobManager 是一个 Flink 集群中任务管理和调度的核心，是控制应用执行的主进程。每个应用都应该被唯一的 JobManager 所控制执行。

JobManger 又包含 3 个不同的组件。

（1）JobMaster

JobMaster 是 JobManager 中最核心的组件，负责处理单独的作业（Job）。JobMaster和具体的 Job 是一一对应的，多个 Job 可以同时运行在一个 Flink 集群中，每个Job都有一个自己的 JobMaster。

1. 在作业提交时，JobMaster 会先接收到要执行的应用。JobMaster 会把JobGraph转换成一个物理层面的数据流图，这个图被叫作“执行图”（ExecutionGraph），它包含了所有可以并发执行的任务。
2. JobMaster 会向资源管理器（ResourceManager）发出请求，申请执行任务必要的资源。一旦它获取到了足够的资源，就会将执行图分发到真正运行它们的TaskManager上。

而在运行过程中 ， JobMaster 会负责所有需要中央协调的操作。

（2）资源管理器（ResourceManager）

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

（3）分发器（Dispatcher）

Dispatcher 主要负责提供一个 REST 接口，用来提交应用，并且负责为每一个新提交的作业启动一个新的 JobMaster 组件。Dispatcher 也会启动一个 Web UI，用来方便地展示和监控作业执行的信息。

2）任务管理器（TaskManager）

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

启动之后，TaskManager 会向资源管理器注册它的 slots；收到资源管理器的指令后，TaskManager 就会将一个或者多个槽位提供给 JobMaster 调用，JobMaster 就可以分配任务来执行了。 在执行过程中，TaskManager 可以缓冲数据，还可以跟其他运行同一应用的TaskManager交换数据。

核心概念

1）并行度（Parallelism）

（1）并行子任务和并行度

Flink 执行过程中，每一个算子（operator）可以包含一个或多个子任务（operatorsubtask），这些子任务在不同的线程、不同的物理机或不同的容器中完全独立地执行。

一个特定算子的子任务（subtask）的个数被称之为其并行度（parallelism）。包含并行子任务的数据流，就是并行数据流。

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

上图数据流中有 source、map、window、sink 四个算子，其中sink算子的并行度为 1，其他算子的并行度都为 2。所以这段流处理程序的并行度就是2。

（2）并行度的设置

在 Flink 中，可以用不同的方法来设置并行度，它们的有效范围和优先级别也是不同的。

• 直接代码设置

代码中在算子后跟着调用 setParallelism()方法，来设置当前算子的并行度：

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

调用执行环境的 setParallelism()方法，全局设定并行度：

env.setParallelism(2);

这样代码中所有算子，默认的并行度就都为 2 。

调试可以使用本地环境，自带Web UI便于调试（默认8081端口）：

需要添加如下依赖：

        <dependency><groupId>org.apache.flink</groupId><artifactId>flink-runtime-web</artifactId><version>${flink.version}</version><scope>provided</scope></dependency>

注意：keyBy 不是算子，所以无法对 keyBy 设置并行度。

• 提交应用时设置

-p 参数来指定当前应用程序执行的并行度，它的作用类似于执行环境的全局设置：

bin/flink run -m 192.168.197.130:8081 -p 2 -c com.lkl.StreamSocketWordCount ../day5-flink-1.0-SNAPSHOT.jar

在 Web UI 上提交作业，也可以在对应输入框中直接添加并行度。

• 配置文件中设置

直接在集群的配置文件 flink-conf.yaml 中直接更改默认并行度：

parallelism.default: 2

初始值为1。

在没有指定并行度的时候，就会采用配置文件中的集群默认并行度。在开发环境中，没有配置文件，默认并行度就是当前机器的CPU核心数。

并行度设置优先级：

代码：算子 > 代码：env > 提交时指定 > 配置文件

2）算子链（Operator Chain）

（1）算子间的数据传输

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

1. 一对一（One-to-one，forwarding）

图中的source 和map算子，source 算子读取数据之后，可以直接发送给 map 算子做处理，它们之间不需要重新分区，也不需要调整数据的顺序。map 算子的子任务，看到的元素个数和顺序跟source算子的子任务产生的完全一样，保证着“一对一”的关系。map、filter、flatMap等算子都是这种one-to-one的对应关系。这种关系类似于 Spark 中的窄依赖。

2. 重分区（Redistributing）

这种模式，数据流的分区会发生改变。比如图中的map和后面的keyBy/window算子之间，以及 keyBy/window 算子和 Sink 算子之间，都是这样的关系。

每一个算子的子任务，会根据数据传输的策略，把数据发送到不同的下游目标任务。这些传输方式都会引起重分区的过程，这一过程类似于 Spark 中的shuffle。

（2）合并算子链

并行度相同的一对一（one to one）算子操作，可以直接链接在一起形成一个“大”的任务（task），这样原来的算子就成为了真正任务里的一部分，如下图所示。每个task 会被一个线程执行。这样的技术被称为“算子链”（Operator Chain）。

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

Flink 默认会按照算子链的原则进行链接合并，如果想要禁止合并或者自行定义，也可以在代码中对算子做一些特定的设置：

// 禁用算子链
.map(word -> Tuple2.of(word, 1L)).disableChaining();
// 从当前算子开始新链
.map(word -> Tuple2.of(word, 1L)).startNewChain()

禁用算子链，当前算子与前后算子都不能组成算子链。（一般不会禁用）

3）任务槽（Task Slots）

（1）任务槽（Task Slots）

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

TaskManager 的计算资源是有限的，并行的任务越多，每个线程的资源就会越少。

为了控制并发量，需要在TaskManager 上对每个任务运行所占用的资源做出明确的划分，这就是所谓的任务槽（taskslots）。

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

独立内存不独立CPU（时间片轮转）

（2）任务槽数量的设置

在 Flink 的 flink-conf.yaml 配置文件中，可以设置TaskManager 的 slot 数量，默认是 1 个 slot。

taskmanager.numberOfTaskSlots: 8

slot 目前仅仅用来隔离内存，不会涉及 CPU 的隔离。在具体应用时，可以将 slot 数量配置为机器的 CPU 核心数，尽量避免不同任务之间对CPU的竞争。

（3）任务对任务槽的共享

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

对于第一个任务节点source→map，它的 6 个并行子任务必须分到不同的 slot 上，而第二个任务节点keyBy/window/apply的并行子任务却可以和第一个任务节点共享 slot。

将资源密集型和非密集型的任务同时放到一个 slot 中，它们就可以自行分配对资源占用的比例，从而保证最重的活平均分配给所有的 TaskManager。

slot 共享另一个好处就是允许保存完整的作业管道。即使某个TaskManager 出现故障宕机，其他节点也可以完全不受影响，作业的任务可以继续执行。

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

.map(word -> Tuple2.of(word, 1L)).slotSharingGroup("aaa");

只有属于同一个 slot 共享组的子任务，才会开启 slot 共享；不同组之间的任务是完全隔离的，必须分配到不同的 slot 上。

4）任务槽和并行度的关系

任务槽是静态的概念 ， 是指 TaskManager 具有的并发执行能力，可以通过参数taskmanager.numberOfTaskSlots 进行配置；而并行度是动态概念，也就是TaskManager 运行程序时实际使用的并发能力，可以通过参数 parallelism.default 进行配置。

注意：

slot数量 >= job并行度（算子最大并行度），job才能运行【否则无法提交任务】

如果是yarn模式，动态申请

​ 申请TM数量 = job并行度 / 每个TM的slot数，向上取整

​ 例如session模式，刚开始0个TaskMananger（设置为3个slot），0个slot

​ 此时，提交一个job并行度为10

​ 10 / 3 向上取整，申请4个TM，使用10个slot，还剩余2个slot