一.部署模式
1.1会话模式(Session Mode)
需要先启动一个 Flink 集群,保持一个会话,所有提交的作业都会运行在此集群上,且启动时所需的资源以确定,无法更改,所以所有已提交的作业都会竞争集群中的资源。
1.2单作业模式(Per-Job Mode)
会为每一个作业单独启动一个集群,且作业完成或关闭后,相应的资源也会被释放。
1.3应用模式(Application Mode)
前两种模式下,应用代码都是在客户端上执行的,然后再由客户端交由 JobManager,客户端需要下载依赖,再将二进制数据发送给 JobManager ,这会加重客户端所在节点的资源压力,而且会占用大量的网络带宽。
所以我们就可以直接将应用提交到 JobManager ,而不通过客户端。引用模式与单作业模式类似,只不过将应用直接交给 JboManager 执行。
二.Flink运行时架构
2.1 作业管理器(JobManager)
JobManager是一个Flink集群中任务管理和调度的核心,是控制应用执行的主进程。且每个应用都i一个被一个唯一的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)。