(增加细粒度资源管理)深入理解flink的task slot相关概念_flink taskmanager task slot

我们知道，flink中每个TaskManager都是一个 JVM 进程，可以在单独的线程中执行一个或多个 subtask(线程)。但是TaskManager 的计算资源是有限的，并不是所有任务都可以放在同一个 TaskManager 上并行执行。并行的任务越多，每个线程的资源就会越少。为了控制并发量，即限制一个 TaskManager 能同时接受多少个 task，我们需要在 TaskManager 上对每个任务运行所占用的资源做出明确的划分，这就是所谓的task slot(任务槽)。

Task Slot是什么

每个（task slot）其实表示了 TaskManager 拥有计算资源的一个固定大小的子集。 这些资源就是用来独立执行一个subtask的。因此，ResourceManager在做资源分配管理的时候，最小的单位就是slot。

具体来说，有 n 个 slot 的 TaskManager，会将其托管内存 1/n 用于每个 slot。分配资源意味着 subtask 不会与其他作业的 subtask 竞争内存,内存是隔离的。

注意此处没有 CPU 隔离，因此不同slot中的subtask会共享CPU。

通过调整 task slot 的数量，用户可以定义 subtask 如何互相隔离。如果每个 TaskManager 只有一个 slot，这意味着这个slot里面的 task 组将独享 JVM 。如果TaskManager包含多个Slot，那么同一个TaskManager中多个Slot内的task可以共享JVM资源，比如共享TCP连接（通过多路复用）、心跳信息、部分数据结构等，从而减少了每个 task 的开销。

官方建议将Slot数目设置为TaskManager下可用的CPU核心数，那么平均下来，每个Slot都能获得1个CPU核心（因为slot只隔离内存不隔离CPU，所以无法强制每个slot独占某个cpu核心）。但考虑到超线程，可以让slotNumber=2*cpuCore.

比如我们有2个 TaskManager  ，每个TaskManager 设置了3个slot，那么slot总数是6 ，在执行任务的时候，如下图：

Task Slot 有哪些资源？

Task Manager 中有固定数量的 Slot ，Slot 的具体数量由配置决定。同一 Task Manager 上 Slot 之间没有差别，每一个 Slot 都一样大，即资源一样多。

每个 slot 都能跑由多个连续 task 组成的一个 pipeline，比如 MapFunction 的第n个并行实例和 ReduceFunction 的第n个并行实例可以组成一个 pipeline。通过动态的对槽的大小和数量的调整，就可以把任务的执行较好的并行起来。

slot隔离主要是对内存的隔离，CPU本身是不做隔离的，CPU在不同的slot之间是可以共享的。

slot内存是平均分配的，比如机器上有16G内存，如果划分4个slot的话，那每个slot就是4G内存了。如果每个slot内存太小的话，任务就执行不下去了，内存直接被撑爆这个是完全有可能的。

JobManager拿到任务执行计划后，它如何确定到底需要多少个slot：只要看整个作业里面，并行度最高的那个算子设置的并行度就可以了，只要满足它的需求，别的就都能满足了。

为了更高效地使用资源,Flink默认允许同一个Job中不同Task的SubTask运行在同一个Slot中，这就是SlotSharing，例如一个作业从Source到Sink的所有子任务都可以放置在一个Slot中，这样数据交换成本更低。

为什么SlotSharing可以提升资源利用率呢？因为不同subtask的内存使用量是不同的，它们被分配到不同的slot后，会导致有的slot内存使用大，有的slot内存使用小。如果不允许slotsharding，slot资源就没有得到充分的利用。

但是slotsharding注意以下描述中的几个关键条件：

1.必须是同一个Job。slot是给Job分配的资源，目的就是隔离各个Job，如果跨Job共享，隔离就失效了；

2.必须是不同Task的Subtask，其实就是让不同操作共享资源,让资源消耗尽可能平均。

3.默认是允许sharing的。

举个例子，下图中两个TaskManager节点共有6个slot，5个SubTask，其中sink的并行度为1，另外两个SubTask的并行度为2。此时由于Subtask（虚线圆角框，5个）少于Slot个数（6个），所以每个Subtask独占一个Slot，没有SlotSharing（想想也是，如果此时还进行slotsharing，就是明明有空余的slot，却让多少subtask去挤破头抢同一个slot的资源，不合理）

下面我们把并行度改为6，此时如果不支持slot sharding,即仍然让每个subtask独占一个slot，那么会导致的问题是：执行速度很慢的subtask会长时间占用一个slot，导致这个slot里面本来空出来的资源没有被其他subtask使用到（这里假设一个slot资源是大于一个subtask所需要的资源）。

因此，如下图，Subtask的个数（虚线圆角框，13个）多于Slot了（6个），出现了SlotSharing。一个Slot中分配了多个Subtask，特别是最左边的Slot中跑了一个完整的Pipeline。

SlotSharing除了提高了资源利用率，还简化了并行度和Slot之间的关系：一个Job运行需要的最少的Slot个数就是其中并行度最大的那个Task（如上图的source+map组合的task）的并行度（即6）,因为并行度最大的task对应的subtask会有6个，而且这6个subtask是不能放在同一个slot里的，所以至少需要6个slot。

Slot Sharing Group(SSG)

为什么又有一个ssg的概念呢？原因有二。

原因一：在 Flink 中，默认情况下，所有的任务都属于同一个默认的 Slot Sharing Group--default。这意味着所有任务都会尝试占用 TaskManager 上的所有可用 slot。这种方式可以在不需要特别考虑 slot 分配的情况下快速启动应用程序，但可能会导致某些任务无法同时在同一个 TaskManager 上运行，从而影响整个应用程序的性能。

为了优化应用程序的性能，我们可以通过设置不同的 Slot Sharing Group 来控制任务之间的共享关系。例如，我们可以将一些任务分组到同一个 Slot Sharing Group 中，让它们共享同一个 TaskManager 上的 slot（CoLocationGroup可以让指定的多个task运行在同一个slot里面，比如迭代计算中）。

原因二：为了防止同一个 slot 包含太多的 task，或者我们希望把计算逻辑复杂的算子单独使用 slot ，提高计算速度。

其实，SSG就是对 operator 进行分组，同一个 group 的不同 operator task 可以共享同一个 slot(相同operator task是不能共享同一个slot的，因为需要保证资源利用率尽量高）。

要想确定一个未做SlotSharingGroup设置的算子的group是什么，可以根据上游算子的 group 和自身是否设置 group共同确定（也就是说如果下游算子没有设置分组，它继承上游算子的分组）； 

只有属于同一个slot共享组的子任务，才会开启slot共享；不同组之间的任务是完全隔离的，必须分配到不同的slot上。在这种场景下，总共需要的slot数量，就是各个slot共享组最大并行度的总和。例如，如果不设置SlotSharingGroup，默认所有task在同一个共享组（可以共享所有slot），那么Flink集群需要的任务槽与作业中使用的最高并行度正好相同。但是如下图所示，如果我们强制指定了map的slot共享组为test，那么map和map下游的组为test，map的上游source的共享组为默认的default，此时default组中最大并行度为10，test组中最大并行度为20，那么需要的Slot=10+20=30；

怎么配置SSG

代码配置

我们可以通过  slotSharingGroup() 为不同的 operator 设置不同的group，决定哪些 task 需要被共享：

还可以通过 CoLocationGroup 来决定哪些 task 需要被单独的 slot 使用，CoLocationGroup可以强制将subTasksk放到同一个slot中，是一种硬约束，但是需要注意两点：

1. 保证把JobVertices的第n个运行实例和其他相同组内的JobVertices第n个实例运作在相同的slot中（所有的并行度相同的subTasks运行在同一个slot ）；
2. 主要用于迭代流(训练机器学习模型) ，用来保证迭代头与迭代尾的第i个subtask能被调度到同一个TaskManager上。

页面配置

 阿里云中还可以为不同group的slot设置不同的资源：作业资源配置 - 实时计算Flink版 - 阿里云

Task Slot与parallelism

一个分配Slot的例子

Flink在调度任务分配Slot的时候遵循两个重要原则：

1. 同一个Job中的同一分组中的不同Task可以共享同一个Slot；
2. Flink是按照拓扑顺序依次从Source调度到sink。

• 假设有两个TM：TM1、TM2，每个TM有3个Slot：S1，S2，S3。假设source/map的并行度为2，keyBy/window/sink的并行度为4，那么调度的顺序依次为source/map[1] ->source/map[2] ->keyBy/window/sink[1]->keyBy/window/sink[2]->keyBy/window/sink[3]->keyBy/window/sink[4]。那么Flink调度任务时（使用默认共享分组）：

1. 首先调度子任务source/map[1]到TM1.S1；
2. 然后调度子任务source/map[2] ，根据Flink的调度原则：source/map[1] 和source/map[2] 属于同一个Task下的两个SubTask，所以他们不能放到同一个Slot中，所以source/map[2]被调度到TM1.S2；
3. 然后调度keyBy/window/sink，keyBy/window/sink的子任务会被依次调度到TM1.S1、TM1.S2、TM2.S1、TM2.S2（之所以不分配到TM1.S3，是为了平衡分配到tm1和tm2）。但是如果source/map与keyBy/window/sink属于不同分组，那么keyBy/window/sink会被调度到TM1.S3、TM2.S1、TM2.S2、TM2.S3。