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

水深无声 2024-02-23 08:16 124阅读 0赞

#### 文章目录 ####

*  1、并行度
 *  2、并行度的设置
 *  3、算子链
 *  4、禁用算子链
 *  5、任务槽
 *  6、任务槽和并行度的关系

## 1、并行度 ##

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

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

![在这里插入图片描述][b82cb02b44f4432bbb85bc7eead57ad1.png]

**某一个算子的子任务的个数被称之为其并行度（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控制台设置一个意思：

![在这里插入图片描述][7af304a5deff48fea6689bcc5ccfdfc6.png]

> 方式三：配置文件中设置

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

parallelism.default: 2

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

![在这里插入图片描述][1aff8eb4f0fa44cfa72b37fe2924e22c.png]

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

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

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

![在这里插入图片描述][769782e33b5b45ddb8a193d9cbc58724.png]  
最后，这几种方式的优先级为：**代码中为某算子单独设定 > 代码中执行环境对象全局设置 > 提交时指定 > 配置文件**

## 3、算子链 ##

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

![在这里插入图片描述][b82cb02b44f4432bbb85bc7eead57ad1.png]

> 一对一（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的直流关系

![在这里插入图片描述][9752b820d48841d58874384d2d51eb9c.png]

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

## 4、禁用算子链 ##

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

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

全局禁用算子链：

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

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

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

![在这里插入图片描述][3a9fbba8a2714385a63ca557bc2f5d4a.png]

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

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

## 5、任务槽 ##

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

![在这里插入图片描述][7f40989f629d4c6586590a386a6439c6.png]

比如一个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个插槽，每个插槽上的子任务对应的算子种类都不一样。

![在这里插入图片描述][00a224a6eb6d496790a91d0aa96dff2b.png]

如上图所示，只要`属于同一个作业`，那么对于`不同任务节点（算子）的并行子任务，就可以放到同一个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

![在这里插入图片描述][69fa7a5454cc416d94b36df2102d5ca0.png]

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

Case2：全局并行度为2

![在这里插入图片描述][eddfd76390064cbd86aba9e5ee6e127c.png]

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

Case3：全局并行度为9

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

![在这里插入图片描述][754153af5d074319be49a9f55dc7940b.png]

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

![在这里插入图片描述][c0b31a3c8c0444adabb260f407f2fa6c.png]

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

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

[b82cb02b44f4432bbb85bc7eead57ad1.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/02/23/a3319df9e8ab4345a9297f1042964a6c.png
[7af304a5deff48fea6689bcc5ccfdfc6.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/02/23/f5655cb398a342549c44786faf16bca4.png
[1aff8eb4f0fa44cfa72b37fe2924e22c.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/02/23/0b15a4da83af417d93929bdd47e2a568.png
[769782e33b5b45ddb8a193d9cbc58724.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/02/23/a003c84147c74ced8c56c05b9a0f027b.png
[9752b820d48841d58874384d2d51eb9c.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/02/23/675d08a6c0fa41198a16ba93b746ed36.png
[3a9fbba8a2714385a63ca557bc2f5d4a.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/02/23/24da9f6981c640fc846bc03719c3e85b.png
[7f40989f629d4c6586590a386a6439c6.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/02/23/17609615ae6e4608a19138cd3304277e.png
[00a224a6eb6d496790a91d0aa96dff2b.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/02/23/ec5285785fcb425db83c166ee1e61f2e.png
[69fa7a5454cc416d94b36df2102d5ca0.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/02/23/800f9017e9f3410e9e3f22f164830c24.png
[eddfd76390064cbd86aba9e5ee6e127c.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/02/23/db4a8d986c944d88ab94520b56f52692.png
[754153af5d074319be49a9f55dc7940b.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/02/23/ca14a8f7084d41809c0a43a0c477f26d.png
[c0b31a3c8c0444adabb260f407f2fa6c.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/02/23/79f346009c2d40efaec8a42b1c2b856b.png