。

也就是说#xff0c;每个应用都应该被唯一的JobManager所控制执行。

作业管理器JobManager

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

也就是说每个应用都应该被唯一的JobManager所控制执行。

1JobMaster

JobMaster是JobManager中最核心的组件负责处理单独的作业Job。

所以JobMaster和具体的Job是一一对应的多个Job可以同时运行在一个Flink集群中,

每个Job都有一个自己的JobMaster。

需要注意在早期版本的Flink中没有JobMaster的概念而JobManager的概念范围较小实际指的就是现在所说的JobMaster。

在作业提交时JobMaster会先接收到要执行的应用。

JobMaster会把JobGraph转换成一个物理层面的数据流图这个图被叫作“执行图”ExecutionGraph它包含了所有可以并发执行的任务。

JobMaster会向资源管理器ResourceManager发出请求申请执行任务必要的资源。

一旦它获取到了足够的资源就会将执行图分发到真正运行它们的TaskManager上。

而在运行过程中JobMaster会负责所有需要中央协调的操作比如说检查点checkpoints的协调。

ResourceManager主要负责资源的分配和管理在Flink

集群中只有一个。

所谓“资源”主要是指TaskManager的任务槽task

slots。

任务槽就是Flink集群中的资源调配单元包含了机器用来执行计算的一组CPU和内存资源。

每一个任务Task都需要分配到一个slot上执行。

这里注意要把Flink内置的ResourceManager和其他资源管理平台比如YARN的ResourceManager区分开。

3分发器Dispatcher

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

UI用来方便地展示和监控作业执行的信息。

Dispatcher在架构中并不是必需的在不同的部署模式下可能会被忽略掉。

2任务管理器TaskManager

TaskManager是Flink中的工作进程数据流的具体计算就是它来做的。

Flink集群中必须至少有一个TaskManager每一个TaskManager都包含了一定数量的任务槽task

slots。

Slot是资源调度的最小单位slot的数量限制了TaskManager能够并行处理的任务数量。

启动之后TaskManager会向资源管理器注册它的slots收到资源管理器的指令后TaskManager就会将一个或者多个槽位提供给JobMaster调用JobMaster就可以分配任务来执行了。

在执行过程中TaskManager可以缓冲数据还可以跟其他运行同一应用的TaskManager交换数据。

二、核心概念

当要处理的数据量非常大时我们可以把一个算子操作“复制”多份到多个节点数据来了之后就可以到其中任意一个执行。

这样一来一个算子任务就被拆分成了多个并行的“子任务”subtasks再将它们分发到不同节点就真正实现了并行计算。

在Flink执行过程中每一个算子operator可以包含一个或多个子任务operator

subtask这些子任务在不同的线程、不同的物理机或不同的容器中完全独立地执行。

一个特定算子的子任务subtask的个数被称之为其并行度parallelism。

这样包含并行子任务的数据流就是并行数据流它需要多个分区stream

partition来分配并行任务。

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

一个程序中不同的算子可能具有不同的并行度。

例如如上图所示当前数据流中有source、map、window、sink四个算子其中sink算子的并行度为1其他算子的并行度都为2。

所以这段流处理程序的并行度就是2。

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

1代码中设置

我们在代码中可以很简单地在算子后跟着调用setParallelism()方法来设置当前算子的并行度

stream.map(word

另外我们也可以直接调用执行环境的setParallelism()方法全局设定并行度

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

我们一般不会在程序中设置全局并行度因为如果在程序中对全局并行度进行硬编码会导致无法动态扩容。

这里要注意的是由于keyBy不是算子所以无法对keyBy设置并行度。

2提交应用时设置

run命令提交应用时可以增加-p参数来指定当前应用程序执行的并行度它的作用类似于执行环境的全局设置

bin/flink

com.atguigu.wc.SocketStreamWordCount

./FlinkTutorial-1.0-SNAPSHOT.jar

如果我们直接在Web

我们还可以直接在集群的配置文件flink-conf.yaml中直接更改默认并行度

parallelism.default:

这个设置对于整个集群上提交的所有作业有效初始值为1。

无论在代码中设置、还是提交时的-p参数都不是必须的所以在没有指定并行度的时候就会采用配置文件中的集群默认并行度。

在开发环境中没有配置文件默认并行度就是当前机器的CPU核心数。

2.2

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

这种模式下数据流维护着分区以及元素的顺序。

比如图中的source和map算子source算子读取数据之后可以直接发送给map算子做处理它们之间不需要重新分区也不需要调整数据的顺序。

这就意味着map

算子的子任务产生的完全一样保证着“一对一”的关系。

map、filter、flatMap等算子都是这种one-to-one的对应关系。

这种关系类似于Spark中的窄依赖。

2重分区Redistributing

在这种模式下数据流的分区会发生改变。

比如图中的map和后面的keyBy/window算子之间以及keyBy/window算子和Sink算子之间都是这样的关系。

每一个算子的子任务会根据数据传输的策略把数据发送到不同的下游目标任务。

这些传输方式都会引起重分区的过程这一过程类似于Spark中的shuffle。

2合并算子链

one算子操作可以直接链接在一起形成一个“大”的任务task这样原来的算子就成为了真正任务里的一部分如下图所示。

每个task会被一个线程执行。

这样的技术被称为“算子链”Operator

上图中Source和map之间满足了算子链的要求所以可以直接合并在一起形成了一个任务因为并行度为2所以合并后的任务也有两个并行子任务。

这样这个数据流图所表示的作业最终会有5个任务由5个线程并行执行。

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

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

禁用算子链.map(word

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

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

那一个TaskManager到底能并行处理多少个任务呢为了控制并发量我们需要在TaskManager上对每个任务运行所占用的资源做出明确的划分这就是所谓的任务槽task

每个任务槽task

slot其实表示了TaskManager拥有计算资源的一个固定大小的子集。

这些资源就是用来独立执行一个子任务的。

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

需要注意的是slot目前仅仅用来隔离内存不会涉及CPU的隔离。

在具体应用时可以将slot数量配置为机器的CPU核心数尽量避免不同任务之间对CPU的竞争。

这也是开发环境默认并行度设为机器CPU数量的原因。

默认情况下Flink是允许子任务共享slot的。

如果我们保持sink任务并行度为1不变而作业提交时设置全局并行度为6那么前两个任务节点就会各自有6个并行子任务整个流处理程序则有13个子任务。

如上图所示只要属于同一个作业那么对于不同任务节点算子的并行子任务就可以放到同一个slot上执行。

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

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

slot共享另一个好处就是允许我们保存完整的作业管道。

这样一来即使某个TaskManager出现故障宕机其他节点也可以完全不受影响作业的任务可以继续执行。

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

.map(word

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

在这种场景下总共需要的slot数量就是各个slot共享组最大并行度的总和。

2.4

任务槽和并行度都跟程序的并行执行有关但两者是完全不同的概念。

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

举例说明假设一共有3个TaskManager每一个TaskManager中的slot数量设置为3个那么一共有9个task

而我们定义word

当所有算子并行度相同时容易看出source和flatmap可以合并算子链于是最终有三个任务节点。

通过这个例子也可以明确地看到整个流处理程序的并行度就应该是所有算子并行度中最大的那个这代表了运行程序需要的slot数量。

三、作业提交流程

API编写的代码生成的最初的DAG图用来表示程序的拓扑结构。

这一步一般在客户端完成。

2作业图JobGraph

StreamGraph经过优化后生成的就是作业图JobGraph这是提交给

JobManager

的数据结构确定了当前作业中所有任务的划分。

主要的优化为将多个符合条件的节点链接在一起合并成一个任务节点形成算子链这样可以减少数据交换的消耗。

JobGraph一般也是在客户端生成的在作业提交时传递给JobMaster。

3执行图ExecutionGraph

JobMaster收到JobGraph后会根据它来生成执行图ExecutionGraph。

ExecutionGraph是JobGraph的并行化版本是调度层最核心的数据结构。

与JobGraph最大的区别就是按照并行度对并行子任务进行了拆分并明确了任务间数据传输的方式。

4物理图Physical

JobMaster生成执行图后会将它分发给TaskManager各个TaskManager会根据执行图部署任务最终的物理执行过程也会形成一张“图”一般就叫作物理图Physical

物理图主要就是在执行图的基础上进一步确定数据存放的位置和收发的具体方式。

有了物理图TaskManager就可以对传递来的数据进行处理计算了。

3.3