我們知道，flink中每個(gè)TaskManager都是一個(gè)?JVM?進(jìn)程，可以在單獨(dú)的線程中執(zhí)行一個(gè)或多個(gè)?subtask(線程)。但是TaskManager?的計(jì)算資源是有限的，并不是所有任務(wù)都可以放在同一個(gè)?TaskManager?上并行執(zhí)行。并行的任務(wù)越多，每個(gè)線程的資源就會(huì)越少。為了控制并發(fā)量，即限制一個(gè)?TaskManager?能同時(shí)接受多少個(gè)?task，我們需要在?TaskManager?上對(duì)每個(gè)任務(wù)運(yùn)行所占用的資源做出明確的劃分，這就是所謂的task slot(任務(wù)槽)。

Task Slot是什么

每個(gè)（task?slot）其實(shí)表示了?TaskManager?擁有計(jì)算資源的一個(gè)固定大小的子集。?這些資源就是用來(lái)獨(dú)立執(zhí)行一個(gè)subtask的。因此，ResourceManager在做資源分配管理的時(shí)候，最小的單位就是slot。

具體來(lái)說(shuō)，有 n?個(gè)?slot?的?TaskManager，會(huì)將其托管內(nèi)存?1/n?用于每個(gè)?slot。分配資源意味著?subtask?不會(huì)與其他作業(yè)的?subtask?競(jìng)爭(zhēng)內(nèi)存,內(nèi)存是隔離的。

注意此處沒(méi)有?CPU?隔離，因此不同slot中的subtask會(huì)共享CPU。

通過(guò)調(diào)整?task?slot?的數(shù)量，用戶(hù)可以定義?subtask?如何互相隔離。如果每個(gè)?TaskManager?只有一個(gè)?slot，這意味著這個(gè)slot里面的?task?組將獨(dú)享?JVM?。如果TaskManager包含多個(gè)Slot，那么同一個(gè)TaskManager中多個(gè)Slot內(nèi)的task可以共享JVM資源，比如共享TCP連接（通過(guò)多路復(fù)用）、心跳信息、部分?jǐn)?shù)據(jù)結(jié)構(gòu)等，從而減少了每個(gè)?task?的開(kāi)銷(xiāo)。

官方建議將Slot數(shù)目設(shè)置為T(mén)askManager下可用的CPU核心數(shù)，那么平均下來(lái)，每個(gè)Slot都能獲得1個(gè)CPU核心（因?yàn)閟lot只隔離內(nèi)存不隔離CPU，所以無(wú)法強(qiáng)制每個(gè)slot獨(dú)占某個(gè)cpu核心）。但考慮到超線程，可以讓slotNumber=2*cpuCore.

比如我們有2個(gè)?TaskManager? ，每個(gè)TaskManager?設(shè)置了3個(gè)slot，那么slot總數(shù)是6?，在執(zhí)行任務(wù)的時(shí)候，如下圖：

Task Slot?有哪些資源？

Task?Manager?中有固定數(shù)量的?Slot?，Slot?的具體數(shù)量由配置決定。同一?Task?Manager?上?Slot?之間沒(méi)有差別，每一個(gè)?Slot?都一樣大，即資源一樣多。

每個(gè)?slot?都能跑由多個(gè)連續(xù)?task?組成的一個(gè)?pipeline，比如?MapFunction?的第n個(gè)并行實(shí)例和?ReduceFunction?的第n個(gè)并行實(shí)例可以組成一個(gè)?pipeline。通過(guò)動(dòng)態(tài)的對(duì)槽的大小和數(shù)量的調(diào)整，就可以把任務(wù)的執(zhí)行較好的并行起來(lái)。

slot隔離主要是對(duì)內(nèi)存的隔離，CPU本身是不做隔離的，CPU在不同的slot之間是可以共享的。

slot內(nèi)存是平均分配的，比如機(jī)器上有16G內(nèi)存，如果劃分4個(gè)slot的話，那每個(gè)slot就是4G內(nèi)存了。如果每個(gè)slot內(nèi)存太小的話，任務(wù)就執(zhí)行不下去了，內(nèi)存直接被撐爆這個(gè)是完全有可能的。

JobManager拿到任務(wù)執(zhí)行計(jì)劃后，它如何確定到底需要多少個(gè)slot：只要看整個(gè)作業(yè)里面，并行度最高的那個(gè)算子設(shè)置的并行度就可以了，只要滿(mǎn)足它的需求，別的就都能滿(mǎn)足了。

為了更高效地使用資源,Flink默認(rèn)允許同一個(gè)Job中不同Task的SubTask運(yùn)行在同一個(gè)Slot中，這就是SlotSharing，例如一個(gè)作業(yè)從Source到Sink的所有子任務(wù)都可以放置在一個(gè)Slot中，這樣數(shù)據(jù)交換成本更低。

為什么SlotSharing可以提升資源利用率呢？因?yàn)椴煌瑂ubtask的內(nèi)存使用量是不同的，它們被分配到不同的slot后，會(huì)導(dǎo)致有的slot內(nèi)存使用大，有的slot內(nèi)存使用小。如果不允許slotsharding，slot資源就沒(méi)有得到充分的利用。

?

但是slotsharding注意以下描述中的幾個(gè)關(guān)鍵條件：

1.必須是同一個(gè)Job。slot是給Job分配的資源，目的就是隔離各個(gè)Job，如果跨Job共享，隔離就失效了；

2.必須是不同Task的Subtask，其實(shí)就是讓不同操作共享資源,讓資源消耗盡可能平均。

3.默認(rèn)是允許sharing的。

舉個(gè)例子，下圖中兩個(gè)TaskManager節(jié)點(diǎn)共有6個(gè)slot，5個(gè)SubTask，其中sink的并行度為1，另外兩個(gè)SubTask的并行度為2。此時(shí)由于Subtask（虛線圓角框，5個(gè)）少于Slot個(gè)數(shù)（6個(gè)），所以每個(gè)Subtask獨(dú)占一個(gè)Slot，沒(méi)有SlotSharing（想想也是，如果此時(shí)還進(jìn)行slotsharing，就是明明有空余的slot，卻讓多少subtask去擠破頭搶同一個(gè)slot的資源，不合理）

下面我們把并行度改為6，此時(shí)如果不支持slot?sharding,即仍然讓每個(gè)subtask獨(dú)占一個(gè)slot，那么會(huì)導(dǎo)致的問(wèn)題是：執(zhí)行速度很慢的subtask會(huì)長(zhǎng)時(shí)間占用一個(gè)slot，導(dǎo)致這個(gè)slot里面本來(lái)空出來(lái)的資源沒(méi)有被其他subtask使用到（這里假設(shè)一個(gè)slot資源是大于一個(gè)subtask所需要的資源）。

因此，如下圖，Subtask的個(gè)數(shù)（虛線圓角框，13個(gè)）多于Slot了（6個(gè)），出現(xiàn)了SlotSharing。一個(gè)Slot中分配了多個(gè)Subtask，特別是最左邊的Slot中跑了一個(gè)完整的Pipeline。

SlotSharing除了提高了資源利用率，還簡(jiǎn)化了并行度和Slot之間的關(guān)系：一個(gè)Job運(yùn)行需要的最少的Slot個(gè)數(shù)就是其中并行度最大的那個(gè)Task（如上圖的source+map組合的task）的并行度（即6）,因?yàn)椴⑿卸茸畲蟮膖ask對(duì)應(yīng)的subtask會(huì)有6個(gè)，而且這6個(gè)subtask是不能放在同一個(gè)slot里的，所以至少需要6個(gè)slot。

Slot?Sharing?Group(SSG)

為什么又有一個(gè)ssg的概念呢？原因有二。

原因一：在?Flink?中，默認(rèn)情況下，所有的任務(wù)都屬于同一個(gè)默認(rèn)的?Slot?Sharing?Group--default。這意味著所有任務(wù)都會(huì)嘗試占用?TaskManager?上的所有可用?slot。這種方式可以在不需要特別考慮?slot?分配的情況下快速啟動(dòng)應(yīng)用程序，但可能會(huì)導(dǎo)致某些任務(wù)無(wú)法同時(shí)在同一個(gè)?TaskManager?上運(yùn)行，從而影響整個(gè)應(yīng)用程序的性能。

為了優(yōu)化應(yīng)用程序的性能，我們可以通過(guò)設(shè)置不同的?Slot?Sharing?Group?來(lái)控制任務(wù)之間的共享關(guān)系。例如，我們可以將一些任務(wù)分組到同一個(gè)?Slot?Sharing?Group?中，讓它們共享同一個(gè)?TaskManager?上的?slot（CoLocationGroup可以讓指定的多個(gè)task運(yùn)行在同一個(gè)slot里面，比如迭代計(jì)算中）。

原因二：為了防止同一個(gè)?slot?包含太多的?task，或者我們希望把計(jì)算邏輯復(fù)雜的算子單獨(dú)使用?slot?，提高計(jì)算速度。

其實(shí)，SSG就是對(duì)?operator?進(jìn)行分組，同一個(gè) group?的不同?operator?task?可以共享同一個(gè)?slot(相同operator?task是不能共享同一個(gè)slot的，因?yàn)樾枰ＷC資源利用率盡量高）。

要想確定一個(gè)未做SlotSharingGroup設(shè)置的算子的group是什么，可以根據(jù)上游算子的?group?和自身是否設(shè)置?group共同確定（也就是說(shuō)如果下游算子沒(méi)有設(shè)置分組，它繼承上游算子的分組）；?

只有屬于同一個(gè)slot共享組的子任務(wù)，才會(huì)開(kāi)啟slot共享；不同組之間的任務(wù)是完全隔離的，必須分配到不同的slot上。在這種場(chǎng)景下，總共需要的slot數(shù)量，就是各個(gè)slot共享組最大并行度的總和。例如，如果不設(shè)置SlotSharingGroup，默認(rèn)所有task在同一個(gè)共享組（可以共享所有slot），那么Flink集群需要的任務(wù)槽與作業(yè)中使用的最高并行度正好相同。但是如下圖所示，如果我們強(qiáng)制指定了map的slot共享組為test，那么map和map下游的組為test，map的上游source的共享組為默認(rèn)的default，此時(shí)default組中最大并行度為10，test組中最大并行度為20，那么需要的Slot=10+20=30；

怎么配置SSG

代碼配置

我們可以通過(guò)??slotSharingGroup()?為不同的?operator?設(shè)置不同的group，決定哪些?task?需要被共享：

還可以通過(guò)?CoLocationGroup?來(lái)決定哪些?task?需要被單獨(dú)的?slot?使用，CoLocationGroup可以強(qiáng)制將subTasksk放到同一個(gè)slot中，是一種硬約束，但是需要注意兩點(diǎn)：

1. 保證把JobVertices的第n個(gè)運(yùn)行實(shí)例和其他相同組內(nèi)的JobVertices第n個(gè)實(shí)例運(yùn)作在相同的slot中（所有的并行度相同的subTasks運(yùn)行在同一個(gè)slot?）；
2. 主要用于迭代流(訓(xùn)練機(jī)?學(xué)習(xí)模型)?，用來(lái)保證迭代頭與迭代尾的第i個(gè)subtask能被調(diào)度到同一個(gè)TaskManager上。

頁(yè)面配置

?阿里云中還可以為不同group的slot設(shè)置不同的資源：作業(yè)資源配置 - 實(shí)時(shí)計(jì)算Flink版 - 阿里云

Task Slot與parallelism

一個(gè)分配Slot的例子

Flink在調(diào)度任務(wù)分配Slot的時(shí)候遵循兩個(gè)重要原則：

1. 同一個(gè)Job中的同一分組中的不同Task可以共享同一個(gè)Slot；
2. Flink是按照拓?fù)漤樞蛞来螐腟ource調(diào)度到sink。

• 假設(shè)有兩個(gè)TM：TM1、TM2，每個(gè)TM有3個(gè)Slot：S1，S2，S3。假設(shè)source/map的并行度為2，keyBy/window/sink的并行度為4，那么調(diào)度的順序依次為source/map[1]?->source/map[2]?->keyBy/window/sink[1]->keyBy/window/sink[2]->keyBy/window/sink[3]->keyBy/window/sink[4]。那么Flink調(diào)度任務(wù)時(shí)（使用默認(rèn)共享分組）：

1. 首先調(diào)度子任務(wù)source/map[1]到TM1.S1；
2. 然后調(diào)度子任務(wù)source/map[2]?，根據(jù)Flink的調(diào)度原則：source/map[1]?和source/map[2]?屬于同一個(gè)Task下的兩個(gè)SubTask，所以他們不能放到同一個(gè)Slot中，所以source/map[2]被調(diào)度到TM1.S2；
3. 然后調(diào)度keyBy/window/sink，keyBy/window/sink的子任務(wù)會(huì)被依次調(diào)度到TM1.S1、TM1.S2、TM2.S1、TM2.S2（之所以不分配到TM1.S3，是為了平衡分配到tm1和tm2）。但是如果source/map與keyBy/window/sink屬于不同分組，那么keyBy/window/sink會(huì)被調(diào)度到TM1.S3、TM2.S1、TM2.S2、TM2.S3。