【基础篇】七、Flink核心概念

1、并行度

要处理的数据量很多时，可以把一个算子的操作（比如前面demo里的flatMap、sum），"复制"多份到多个节点，数据来了以后可以到任意一个节点执行。即将一个算子任务拆分成多个并行的子任务，再分发到不同的节点上执行，实现真正的并行计算。（好绕口，就是把一个活儿让好几个Task节点共同去做）

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

某一个算子的子任务的个数被称之为其并行度（parallelism）。 一条流水线上，几个人在同时干着打螺丝，几个人在同时处理着焊电路板。同一个程序，不同的算子，可以有不同的并行度。一个流程序的并行度，可以认为就是其所有算子中最大的并行度。如上图，source、map、window、sink四个算子，sink为1，其余为2，则这个流处理程序的并行度为2。

2、并行度的设置

方式一：代码中设置

算子后跟着调用setParallelism()方法为某一个算子设置并行度

stream.map(word -> Tuple2.of(word, 1L)).setParallelism(2);   //map算子并行度为2

执行环境对象后面调setParallelism()方法设置全局并行度，对所有算子生效

env.setParallelism(2);

一般不设全局，会导致无法动态扩容。

方式二：提交应用时指令中设置

-p参数来指定当前应用程序执行的并行度，类似上面的全局设置

bin/flink run –p 2 –c com.plat.SocketStreamWordCount  ./FlinkDemo-1.0-SNAPSHOT.jar

这种和Web控制台设置一个意思：

方式三：配置文件中设置

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

parallelism.default: 2

这个设置对于整个集群上提交的所有作业有效，初始值为1。当代码中没设置、提交时没指定，就用这个配置文件的。在开发环境中，没有配置文件，默认并行度就是当前机器的CPU核心数。

最后，本地调试想看控制台界面，可创建本地环境执行对象，用于本地调试：

StreamExecutionEnvironment env = StreamExecutionEnvironment.createlocalEnvironmentWithwebuI(new Configuration());

访问localhosr:8081，socket是特殊的，只能是1，改不了，其余算子均为4。

最后，这几种方式的优先级为：代码中为某算子单独设定 > 代码中执行环境对象全局设置 > 提交时指定 > 配置文件

3、算子链

一个数据流，数据在各种算子之间传输的形式可能是一对一（one-to-one）的直通（forwarding），也可能是打乱的重分区（redistributing）。

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

如上图，source算子读完数据后，可以直接发给map算子接着处理。map 算子的子任务，看到的元素个数和顺序跟source 算子的子任务产生的完全一样，即一对一，一个算子的task和一个算子的task数据一样。特点是：

• 数据不需要重新分区
• 数据不需要调整顺序

重分区（Redistributing）

和一对一的直流相反，此时数据的分区会发生改变，如图中，map完数据后，直接keyBy/window（注意keyBy自身不是算子），按key分组了。也就是每一个算子的子任务task，会根据某些规则，把数据发送到不同的下游task，从而引起了数据重分区。

合并算子链

在Flink中，并行度相同的一对一（one to one）算子操作，可以直接连接在一起形成一个大的任务（task），每个task又会被一个线程执行，即算子链。合并的条件：

• 两算子并行度相等（子任务数量一样）
• 两算子为one to one的直流关系

上图中Source和map之间满足了算子链的要求，所以可以直接合并在一起，形成了一个任务；因为并行度为2，所以合并后的任务也有两个并行子任务。这样，这个数据流图所表示的作业最终会有5个任务，由5个线程并行执行。合并算子链的机制，可以减少线程之间的切换和基于缓存区的数据交换，在减少时延的同时提升吞吐量。

4、禁用算子链

Flink默认会按照算子链的原则进行链接合并，但有的场景不适合合并，比如：

• 两个算子串在一起，它们的子任务task搭配形成n组（n为并行度），每组共用一个线程，但如果两个算子本身计算任务都很重，那就不适合串一起，就像两个脾气都差的人合租，此时应该断开算子链
• 当出现错误，需要定位问题是哪个算子时，就要禁用算子链

全局禁用算子链：

//env为执行环境对象
env.disableOperatorChaining();

disableChaining方法可只给某个算子设置禁用算子链，那它和它前后的算子就都不能再组成算子链（控制台上UI会显示Forward，表明本来是一对一的算子链关系）

.map(word -> Tuple2.of(word, 1L)).disableChaining();

startNewChain方法，从当前算子开始新链，即只和前面的算子断开，和后面的算子能串一起的话还是会串

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

5、任务槽

Flink中每一个TaskManager都是一个JVM进程，它可以启动多个独立的线程，来并行执行子任务。但每个TaskManager总的计算资源有限，并行任务越多，每个线程能分到的可用资源就越少，为了限制TaskManager能并行处理的最大任务数，提出任务槽（task slots）的概念，对TaskManager上对每个任务运行所占用的资源做出明确的划分。一锅饭，能盛6碗，谁来都夹一筷子，谁都吃不饱，因此，锅前放6个碗，也就是分为6碗饭，来一个人，就端走一碗，端没了别人就去其他锅，分到饭的六个人，也不用和别人抢，且能吃饱。这个碗就是任务槽。

比如一个TaskManager上有三个slot，那就把这个TaskManager的内存资源分为三份，一个插槽一份。如此，在插槽上执行一个子任务时，就相当于划定了一块内存给这个子任务专款专用，不需要和其他子任务去竞争内存资源。前面提到的合并成算子链后的5个子任务，两个TaskManager就可实现，如上图。

任务槽数量的设置

在flink安装目录的conf/flink-conf.yaml配置文件中，可以设置每个TaskManager的slot数量，默认是1个slot。

taskmanager.numberOfTaskSlots: 8

slot目前仅仅用来隔离内存，不会涉及CPU的隔离。在具体应用时，可以将slot数量配置为机器的CPU核心数，尽量避免不同任务之间对CPU的竞争。这也是开发环境默认并行度设为机器CPU数量的原因。（这就像合租，内存就像卧室，厕所就像CPU，三个人，三间房，但一个厕所也够用，类比CPU时间片和线程切换）

子任务task对任务槽的共享

上面讲到，一人一碗饭，一个子任务一个插槽。而插槽的共享，就是放宽了政策，不同类型的算子，它们的并行子任务允许放到同一个插槽上并行执行（注意，依旧并行）。如下图，两个TaskManager，6个插槽，每个插槽上的子任务对应的算子种类都不一样。

如上图所示，只要属于同一个作业，那么对于不同任务节点（算子）的并行子任务，就可以放到同一个slot上执行。所以对于第一个任务节点source→map，它的6个并行子任务必须分到不同的slot上，而第二个任务节点keyBy/window/apply的并行子任务却可以和第一个任务节点共享slot。slot共享的好处在于：

• 活儿大致平均分配到了所有的TaskManager
• slot有好几种算子的子任务，组合起来就是一个完整的作业管道或者流。此时，即使某个TaskManager宕机，其他节点也不受影响，作业继续执行

如果不希望默认的slot共享，比如需要让某个算子的task独享一个slot，就可以设置slot共享组。

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

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

6、任务槽和并行度的关系

• 任务槽slot是一个静态概念，表示最大的并发上限。假设一共有3个TaskManager，每一个TaskManager中的slot数量设置为3个，那么一共有9个task slot，表示集群最多能并行执行同一算子的9个子任务。
• 并行度是一个动态概念，表示实际运行占用了几个。比如并行度为4，即这个算子有4个子任务task，需要放在4个插槽上。此时，并行度为4，slot为9。

Job运行时，必须插槽slot的数量必须大于等于并行度，否则任务运行失败：NoResourceAvailableException:could not acquire the minimun required resources 。注意Yarn等模式部署时，会动态申请TaskManager，申请的TM的数量 = job并行度 /每个TM的slot的数量，向上取整。

比如，某算子并行度为10，即它有10个task要放在不同的插槽上，此时插槽有9个，那就不能运行，而不是9个跑完再让第十个执行。再比如，一个Flink程序中定义了4个算子：

source→ flatmap→ reduce→ sink

前提： flink-conf.yaml中taskmanager.numberOfTaskSlots数量为3（建议为CPU核心数）,假设TaskManager数量也为3，即插槽有3*3=9个

Case1：并行度parallelism.default=1

分析：4种算子，并行度为1 ⇒ 其中两个形成算子链算一个 ⇒ 三个子任务 ⇒ 同一作业的不同种类的算子的任务，共享任务槽 ⇒ 总共占用一个插槽，剩8个可用

Case2：全局并行度为2

分析：三种算子，并行度为2 ⇒ 其中两个形成算子链算一个 ⇒六个子任务 ⇒ 插槽共享 ⇒ 总共占用2个插槽，剩7个可用 ⇒ 计算机资源利用不充分，设置合适的并行度才能提高效率

Case3：全局并行度为9

分析： 并行度为9 ⇒ 一种算子有9个子任务 ⇒ 插槽共享 ⇒ 占九个

Case4：全局set为9，但sink算子set为1

分析： 并行度为9 ⇒ 一种算子有9个子任务 ⇒ 29 + 11 = 19个子任务 ⇒ 插槽共享

最后，可以看到，整个流处理程序的并行度，就是所有算子并行度的最大值，因为这代表了程序运行所需要的插槽slot的数量。