- 1二叉树与哈希表以及基本算法_二叉树和hash表
- 2前端代码规范及最佳实践_前端代码规范文档
- 3Android开发“仿美团”_用安卓写一个有关美团外卖的项目
- 4【AIGC半月报】AIGC大模型启元：2024.04（上）_streamingt2v最新版本更新内容
- 5【Verilog异步清零计数器】
- 6秋招后端开发面试题 - Java多线程（上）
- 7关于GitLab登录/推送/拉取代码时候报错（remote: HTTP Basic: Access deniedfatal: Authentication failed fo ‘xxxx‘.）_remote: http basic: access denied fatal: authentic
- 8c++小游戏合集
- 9【计算机毕设文章】基于Spring Boot的美食分享系统设计与实现_i21mn_美食分享系统数据设计
- 10计算机毕业设计 ssm高校课程教学平台系统B4 毕设
Flink 中 slot ，task，并行度的概念以及与CPU，内存的关系_task slots
赞
踩
我们在学习Flink时，经常会听到 task，slot，线程以及并行度这几个概念，对于初学者来说，这几个概念关系以及它们与系统内存，CPU之间的关系经常搞不清楚，下面我们就通过这篇文章来弄清楚这些概念与关系。
并行度
特定算子的子任务（subtask）的个数称之为并行度（parallel），一般情况下，一个数据流的并行度可以认为是其所有算子中最大的并行度。Flink中每个算子都可以在代码中通过.setParallelism(n)来重新设置并行度。而并行执行的subtask要发布到不同的slot中去执行。
Tasks
对于分布式执行的任务，Flink 将算子的 subtasks 链接成 tasks。每个 subtask 由一个线程执行。
下图中样例数据流用 5 个 subtask 执行，因此有 5 个并行线程。
算子链
将算子链接成 task 是个有用的优化：
- 它减少线程间切换、缓冲的开销，并且减少延迟的同时增加整体吞吐量。
- 链行为是可以配置的；将两个算子链接在一起能使得它们在同一个线程中执行，从而提升性能。
Flink 默认会将能链接的算子尽可能地进行链接(例如， 两个 map 转换操作)。 此外， Flink 还提供了对链接更细粒度控制的 API 以满足更多需求：
如果想对整个作业禁用算子链，可以调用 StreamExecutionEnvironment.disableOperatorChaining()。下列方法还提供了更细粒度的控制。需要注 意的是， 这些方法只能在 DataStream 转换操作后才能被调用，因为它们只对前一次数据转换生效。例如，可以 someStream.map(...).startNewChain() 这样调用，而不能 someStream.startNewChain()这样。
一个资源组对应着 Flink 中的一个 slot 槽， 你可以根据需要手动地将各个算子隔离到不同的 slot 中。
| Transformation | Description |
|---|---|
| Start new chain | 以当前 operator 为起点开始新的连接。如下的两个 mapper 算子会链接在一起而 filter 算子则不会和第一个 mapper 算子进行链接。someStream.filter(...).map(...).startNewChain().map(...); |
| Disable chaining | 任何算子不能和当前算子进行链接; someStream.map(...).disableChaining(); |
| Set slot sharing group | 配置算子的资源组。Flink 会将相同资源组的算子放置到同一个 slot 槽中执行，并将不同资源组的算子分配到不同的 slot 槽中，从而实现 slot 槽隔离。资源组将从输入算子开始继承如果所有输入操作都在同一个资源组。 Flink 默认的资源组名称为 “default”，算子可以显式调用 slotSharingGroup(“default”) 加入到这个资源组中。 someStream.filter(...).slotSharingGroup("name"); |
Task Slots 和资源
每个 worker（TaskManager）都是一个 JVM 进程，可以在单独的线程中执行一个或多个 subtask。为了控制一个 TaskManager 中接受多少个 task，就有了所谓的 task slots（至少一个）。每个 task slot 代表 TaskManager 中资源的固定子集。
例如，具有 3 个 slot 的 TaskManager，会将其托管内存 1/3 用于每个 slot。分配资源意味着 subtask 不会与其他作业的 subtask 竞争托管内存，而是具有一定数量的保留托管内存。注意此处没有 CPU 隔离；当前 slot 仅分离 task 的托管内存。通过调整 task slot 的数量，用户可以定义 subtask 如何互相隔离。每个 TaskManager 有一个 slot，这意味着每个 task 组都在单独的 JVM 中运行（例如，可以在单独的容器中启动）。具有多个 slot 意味着更多 subtask 共享同一 JVM。同一 JVM 中的 task 共享 TCP 连接（通过多路复用）和心跳信息。它们还可以共享数据集和数据结构，从而减少了每个 task 的开销。
上边例子，从图中可以看出来5个subtask应该需要5个task slot来执行，事实是这样的吗？
不是的，默认情况下，上边例子只需要2个slot就可以了。
我们再看另外一个例子，当我们把并行度调大为6：
按照并行度拆开这个任务（task），我们发现会有13个subtask，那么是不是就意味着需要13个slot才能执行该任务呢？
答案是否定的，实际是只需要6个slot就够了。
为什么会这样呢？
我们先来看两条规则：
- 默认情况下，Flink 允许子任务共享slot，即使他们是不同任务的子任务。这样的结果就是一个slot可以保存作业的整个pipeline。
- Task slot 是静态的概念，指的是TaskManager具有的并发执行能力。
从图中可以看出来，第一个slot会运行3个subtask，也就是执行3个线程，前面提到slot只是做了内存隔离，并没有做CPU隔离，假设这样一种情况，我们的服务器是6核CPU的，也就是意味着每个slot就可以分到一个CPU资源，那么就意味着这3个子任务中，一个子任务执行时就有2个子任务在等待状态，所以我们在设置slot个数时，也要考虑一下集群的资源，尽量使得每个slot能使用得到合理的CPU资源。
那么问题又来了，上面这个图中画的需要5个task slot，但是默认情况下Flink会自动优化成为需要2个slot，如果我们就想让这些任务使用5个slot来执行呢，那就要通过禁用算子链来实现了：
我们看如下代码：
// 从socket文本流读取数据
DataStream<String> inputDataStream = env.socketTextStream(host, port);
// 基于数据流进行转换计算
DataStream<Tuple2<String, Integer>> resultStream = inputDataStream.flatMap(new WordCount.MyFlatMapper()).slotSharingGroup("green")
.keyBy(0)
.sum(1).setParallelism(2).slotSharingGroup("red");
resultStream.print().setParallelism(1);
- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
这几行代码几个subtask？并行度是多少？用几个task slot？
以上代码运行时flink webUI：
可以看出，该任务被切分成了5个子task，按照最大并行度算子来算，这个任务的并行度应该为2，那么这5个subtask占用了几个slot呢？
我们先按照算子链进行分组，每个分组最大的并行度相加，就是这个任务所占用的总共slot，所以应该是4个：
总结
- Flink中slot是任务执行所申请资源的最小单元，同一个TaskManager上的所有slot都只是做了内存分离，没有做CPU隔离。
- 每一个TaskManager都是一个JVM进程，如果某个TaskManager 上只有一个 slot，这意味着每个 task 组都在单独的 JVM 中运行，如果有多个 slot 就意味着更多 subtask 共享同一 JVM。
- 一般情况下有多少个subtask，就是有多少个并行线程，而并行执行的subtask要发布到不同的slot中去执行。
- Flink 默认会将能链接的算子尽可能地进行链接，也就是算子链，flink 会将同一个算子链分组内的subtask都发到同一个slot去执行，也就是说一个slot可能要执行多个subtask，即多个线程。
- flink 可以根据需要手动地将各个算子隔离到不同的 slot 中。
- 一个任务所用的总共slot为所有资源隔离组所占用的slot之和，同一个资源隔离组内，按照算子的最大并行度来分配slot。
关注我们：
