一、Flink并行度

1. 并行度的概念

在Flink中，并行度（Parallelism）是指在Flink作業中并行執行任務的程度，它決定了作業中任務的數量以及任務之間的數據劃分和分配方式，是實現高吞吐量和低延遲流處理的關鍵概念。一個特定算子的子任務（subtask）的個數被稱之為其并行度。包含并行子任務的數據流，就是并行數據流，它需要多個分區（stream partition）來分配并行任務。一般情況下，一個流程序的并行度，可以認為就是其所有算子中最大的并行度。一個程序中，不同的算子可能具有不同的并行度。

例如，在一個Flink程序中，Source算子的并行度設置為2，Map算子的并行度設置為4，Sink算子的并行度設置為1，那么這個程序的并行度就是4。

Flink程序本質上是并行的和分布式的，在執行過程中，一個流（stream）包含一個或多個流分區，而每一個operator包含一個或多個operator子任務。操作子任務間彼此獨立，在不同的線程中執行，甚至是在不同的機器或不同的容器上。operator子任務的數量是這一特定operator的并行度。相同程序中的不同operator有不同級別的并行度。

2. 并行度的設置

Flink可以在不同的級別設置并行度，包括算子層次、執行環境層次、客戶端層次和系統層次，且優先級為：算子層次 > 執行環境層次 > 客戶端層次 > 系統層次。

(1) 算子層次

單個算子、數據源和數據接收器的并行度可以通過調用 setParallelism() 方法來指定。這種方式設置的并行度，只針對當前算子有效。

finalStreamExecutionEnvironment env =StreamExecutionEnvironment.getExecutionEnvironment();
DataStream<String> text =[...];
DataStream<Tuple2<String,Integer>> wordCounts = text
.flatMap(newLineSplitter())
.keyBy(value -> value.f0)
.window(TumblingEventTimeWindows.of(Time.seconds(5)))
.sum(1).setParallelism(5);

wordCounts.print();
env.execute("Word Count Example");

在上述代碼中，sum(1) 算子的并行度被設置為5，這意味著該算子會啟動5個并發任務來處理數據。

(2) 執行環境層次

Flink程序運行在執行環境的上下文中，執行環境為所有執行的算子、數據源、數據接收器（sink）定義了一個默認的并行度。可以顯式配置算子層次的并行度去覆蓋執行環境的并行度。可以通過調用 setParallelism() 方法指定執行環境的默認并行度。

finalStreamExecutionEnvironment env =StreamExecutionEnvironment.getExecutionEnvironment();
env.setParallelism(3);

DataStream<String> text =[...];
DataStream<Tuple2<String,Integer>> wordCounts =[...];
wordCounts.print();
env.execute("Word Count Example");

在上述代碼中，執行環境的并行度被設置為3，這意味著如果沒有在算子層次顯式設置并行度，所有算子的并行度都將為3。

(3) 客戶端層次

將作業提交到Flink時可在客戶端設定其并行度。對于CLI客戶端，可以通過 -p 參數指定并行度。

./bin/flink run -p 10 WordCount-java.jar

在上述命令中，作業的并行度被設置為10。

(4) 系統層次

可以通過設置 ./conf/flink-conf.yaml 文件中的 parallelism.default 參數，在系統層次來指定所有執行環境的默認并行度。

parallelism.default:2

在上述配置中，系統默認的并行度被設置為2。

3. 如何合理規劃并行度

(1) 理解任務特性和需求

• 任務類型：CPU密集型任務可能需要較高的并行度來充分利用計算資源，而I/O密集型任務可能需要較低的并行度以減少資源競爭和網絡開銷。例如，對于一個復雜的數據分析任務，可能是CPU密集型的，此時可以適當提高并行度；而對于一個從數據庫讀取數據的任務，可能是I/O密集型的，過高的并行度可能會導致網絡擁塞。
• 數據分布：如果數據分布不均勻，可能會導致某些任務負載過重，影響整體性能。此時，調整并行度可以使數據分布更均勻。比如，在處理用戶行為數據時，可能某些熱門用戶的數據量遠遠大于其他用戶，這時候可以通過調整并行度來避免數據傾斜。

(2) 考慮集群資源限制

• 資源可用性：集群的資源（如CPU核心數、內存大小、網絡帶寬等）會限制可以設置的并行度。需要根據集群的實際情況來合理設置。例如，如果集群的CPU核心數有限，設置過高的并行度會導致任務競爭CPU資源，反而降低性能。
• 資源競爭：過高的并行度可能導致資源競爭加劇，反而降低整體性能。比如，多個任務同時競爭內存資源，可能會導致頻繁的內存交換，影響任務的執行效率。

(3) 分析作業結構和數據流動

• 算子依賴關系：作業中不同算子之間的依賴關系會影響數據流動和并行度的設置。需要確保數據能夠高效地在算子之間傳遞。例如，如果一個算子依賴于另一個算子的輸出結果，那么它們的并行度設置需要相互匹配，以避免數據阻塞。
• 數據傾斜：某些算子可能處理的數據量遠大于其他算子，導致數據傾斜。通過調整并行度可以減少數據傾斜的影響。比如，對于一個聚合算子，如果某些分組的數據量過大，可以適當提高該算子的并行度，將數據分散到更多的任務中處理。

(4) 實際應用中的設置方法

• 算子級并行度：通過調用 setParallelism() 方法可以在算子操作后設置其并行度。這種方法允許對特定算子進行精細控制。例如：

DataStream<String> stream =...;
stream.map(newMyMapFunction()).setParallelism(2);

• 作業級并行度：在創建執行環境后，可以通過調用 setParallelism() 方法設置全局的默認并行度。這種方法適用于對整個作業進行統一配置。例如：

StreamExecutionEnvironment env =StreamExecutionEnvironment.getExecutionEnvironment();
env.setParallelism(4);

• 客戶端設置：在提交任務時，可以通過命令行接口（CLI）的 -p 參數或Java程序中的相應設置來指定并行度。例如：

./bin/flink run -p 10 your-job.jar

• 集群默認設置：在集群的配置文件（如 flink-conf.yaml）中設置默認并行度，這將影響集群上提交的所有作業。例如：

parallelism.default:2

(5) 監控和調整

• 監控執行情況：通過Flink的Web UI或其他監控工具監控作業的執行情況和集群資源利用率。例如，觀察任務的處理延遲、資源使用情況等指標。
• 動態調整：根據實際情況動態調整并行度，以適應不同的工作負載和數據流量。例如，如果發現某個算子的處理延遲過高，可以適當提高其并行度。

(6) 注意事項

• 并行度與性能的關系：并行度并非越高越好，需要根據實際情況進行權衡。過高的并行度可能導致資源競爭和開銷增加，反而降低性能。
• 考慮未來擴展性：在設置并行度時，還需要考慮作業的擴展性和未來可能的需求變化。例如，隨著業務的發展，數據量可能會增加，此時需要預留一定的并行度擴展空間。

4. 不同組件的并行度設置建議

(1) 數據源（Source）

• Kafka數據源：如果數據源是Kafka，Source的并行度通常設置為Kafka對應Topic的分區數。這樣可以確保每個分區的數據由一個獨立的并行任務來處理，充分利用資源。例如，如果Kafka的某個Topic有10個分區，那么Source的并行度可以設置為10。如果消費速度仍跟不上數據生產速度，可以考慮擴大Kafka的分區數，并相應地調大并行度。但要注意，如果并行度多于Kafka的分區數，會造成有的并行度空閑，浪費資源。
• 其他數據源：對于其他數據源，需要根據數據源的特性和數據量來設置并行度。如果數據源支持并行讀取，可以適當提高并行度；如果數據源的讀取性能有限，過高的并行度可能會導致資源競爭和性能下降。

(2) 轉換算子（Transform）

• Keyby之前的算子：如map、filter、flatmap等，這些算子一般不會做太重的操作，并行度可以和Source保持一致，使得算子之間可以做到forward傳輸數據，不經過網絡傳輸，提高處理效率。
• Keyby之后的算子：如果并發較大，建議設置并行度為2的整數次冪，例如128、256、512等。這是因為Flink內部的一些機制（如狀態管理）在并行度為2的整數次冪時能更好地工作，使數據相對均勻地shuffle到下游算子。對于小并發任務，并行度不一定需要設置成2的整數次冪。如果大并發任務沒有KeyBy操作，并行度也無需設置為2的整數次冪。

(3) 數據接收器（Sink）

Sink是數據流向下游的地方，可以根據Sink的數據量及下游的服務抗壓能力進行評估。如果Sink是Kafka，可以設為Kafka對應Topic的分區數，并且Sink并行度最好和Kafka partition成倍數關系，否則可能會出現到Kafka partition數據不均勻的情況。但大多數情況下，Sink算子并行度不需要特別設置，只需要和整個任務的并行度相同就行。如果下游服務的抗壓能力有限，需要適當降低并行度；如果下游服務能夠處理大量數據，可以提高并行度以提高吞吐量。

二、Flink Slot

1. Slot的概念

在Apache Flink中，Task Slot（任務槽）是TaskManager的一個關鍵概念，它用于資源管理和并行任務的調度。每個TaskManager可以擁有多個Task Slot，每個Task Slot能夠獨立地運行一個或多個子任務（subtask）。Slot是計算資源的隔離單元，一個Slot可以運行多個SubTask，但是這些SubTask必須是來自同一個application的不同階段的subTask。

Slot數量通常與每個TaskManager節點的可用CPU內核數成比例，一般Slot數量是每個節點的CPU內核數。Slot的數量由集群中 flink-conf.yml 配置文件中 taskmanager.numberOfTaskSlots 設置。例如，在 flink-conf.yml 中設置 taskmanager.numberOfTaskSlots: 4，表示每個TaskManager有4個Slot。

2. Slot的特性

(1) 并行度與Task Slot數量的關系

每個Task Slot可以執行一個并行任務實例（即一個subtask）。因此，Job的最大并行度受限于所有TaskManager上可用Task Slot的總數。如果Job的并行度大于總Task Slot數，則部分任務將排隊等待空閑的Task Slot；反之，如果Task Slot數量過多而實際并行度較低，則會造成資源浪費。

(2) 資源共享與隔離

在同一個TaskManager內的不同Task Slot之間，網絡連接、文件句柄等非內存資源是共享的，但每個Task Slot有自己獨立的內存空間。這種設計既保證了一定程度上的資源隔離（如避免內存溢出），又允許一定程度的資源共享（如減少網絡開銷），從而提高了整體效率。

(3) 鏈式調度（Chaining）

當多個連續的任務屬于同一算子并且具有相同的并行度時，Flink可以將它們合并到同一個Task Slot中執行，稱為“鏈式調度”。這樣做的好處是可以減少線程切換和序列化/反序列化的開銷。鏈式調度的前提條件是這些任務之間沒有其他類型的任務插入，且它們的操作不會導致阻塞。

(4) 動態調整

用戶可以在配置文件中設置默認的Task Slot數量，也可以在啟動集群時通過命令行參數指定。此外，還可以根據具體作業的需求動態調整每個TaskManager的Task Slot數量。動態調整Task Slot數量的能力有助于更好地適應不同類型的負載變化，例如高峰期增加Task Slot來提升吞吐量，在低谷期減少Task Slot以節省資源。

3. Slot的配置方式

(1) 全局配置

在 flink-conf.yaml 文件中設置 taskmanager.numberOfTaskSlots 參數，為整個集群設定默認的Task Slot數量。

taskmanager.numberOfTaskSlots:4

(2) 命令行參數

在啟動Flink集群時，使用 -D taskmanager.numberOfTaskSlots=4 參數覆蓋默認值。

./bin/start-cluster.sh -D taskmanager.numberOfTaskSlots=4

(3) 動態配置

對于某些特定的應用場景，可能需要更靈活地控制每個TaskManager的Task Slot數量。這時可以利用Flink提供的REST API或者YARN/Kubernetes等平臺提供的機制來進行動態調整。

4. Slot的最佳實踐

(1) 合理規劃并行度

確保Job的并行度與集群中的Task Slot總數相匹配，既能充分利用現有資源，又不會造成不必要的等待。例如，如果集群中有10個TaskManager，每個TaskManager有4個Slot，那么集群總共有40個Slot，Job的并行度可以設置為40以內。

(2) 考慮任務特性

對于計算密集型任務，可以適當增加Task Slot的數量以提高并發處理能力；而對于I/O密集型任務，則應關注網絡帶寬和磁盤I/O性能。例如，對于一個計算密集型的數據分析任務，可以增加Task Slot的數量，讓更多的子任務并行執行。

(3) 監控資源使用情況

定期檢查Task Slot的使用率、內存消耗等指標，及時發現并解決潛在的問題。例如，可以使用Flink的監控工具，查看每個TaskManager的Slot使用率和內存使用情況。

(4) 測試與調優

在生產環境中部署之前，先在小規模集群上進行充分測試，找到最適合當前工作負載的Task Slot配置。例如，可以在測試環境中調整Task Slot的數量，觀察作業的性能變化，找到一個最優的配置。

5. 算子和Slot的關系

(1) 算子的子任務與Slot的分配

在Flink中，每個算子會根據其并行度被拆分成多個子任務（subtask），這些子任務需要被分配到不同的Slot中執行。例如，一個算子的并行度為3，那么它會有3個子任務，這3個子任務需要被分配到3個不同的Slot中。默認情況下，Flink允許子任務共享Slot，只要這些子任務屬于同一作業。這樣可以提高資源利用率，例如將資源密集型和非密集型的任務同時放到一個Slot中，它們可以自行分配對資源占用的比例。

(2) 算子鏈對Slot使用的影響

算子鏈（Operator Chain）是Flink中的一種優化技術，它將多個算子連接在一起形成一個鏈式結構，以減少數據序列化和網絡傳輸開銷。當算子形成算子鏈后，它們會被合并成一個任務，這個任務只需要一個Slot來執行。例如，Source算子和Map算子形成了算子鏈，它們的并行度都為2，那么合并后的任務也有2個子任務，只需要2個Slot來執行。

(3) Slot共享組對算子和Slot關系的影響

通過 slotSharingGroup 方法可以將算子分配到指定的共享組中，同一共享組的算子會盡可能共享Slot。例如，將多個算子都設置為同一個共享組，那么這些算子的子任務可以共享同一個Slot，從而提高資源利用率。但如果不同的算子設置了不同的共享組，它們的子任務就不能共享Slot，需要分別分配Slot。

三、Flink算子鏈

1. 算子鏈的概念

算子鏈（Operator Chain）是Flink中的一種優化技術，用于將多個算子連接在一起形成一個鏈式結構，以減少數據序列化和網絡傳輸開銷，提高整體的處理性能。在Flink中，并行度相同的一對一（one to one）算子操作，可以直接鏈接在一起形成一個 “大” 的任務（task），每個task會被一個線程執行。

例如，在一個WordCount程序中，Source算子和Map算子之間滿足算子鏈的要求，可以直接合并在一起，形成一個任務；因為并行度為2，所以合并后的任務也有兩個并行子任務。這樣，這個數據流圖所表示的作業最終會有5個任務，由5個線程并行執行。

2. 算子間的數據傳輸模式

(1) 一對一（One-to-one，forwarding）

這種模式下，數據流維護著分區以及元素的順序。比如source和map算子，source算子讀取數據之后，可以直接發送給map算子做處理，它們之間不需要重新分區，也不需要調整數據的順序。這就意味著map算子的子任務，看到的元素個數和順序跟source算子的子任務產生的完全一樣，保證著“一對一”的關系。map、filter、flatMap 等算子都是這種one-to-one的對應關系。這種關系類似于Spark中的窄依賴。

(2) 重分區（Redistributing）

在這種模式下，數據流的分區會發生改變。比如 map 和后面的 keyBy/window 算子之間，以及 keyBy/window 算子和 Sink 算子之間，都是這樣的關系。每一個算子的子任務，會根據數據傳輸的策略，把數據發送到不同的下游目標任務。這些傳輸方式都會引起重分區的過程，這一過程類似于Spark中的shuffle。

3. 算子鏈的創建條件

(1) 上下游的并行度一致

上下游算子的并行度必須相同，才能形成算子鏈。例如，如果Source算子的并行度為2，Map算子的并行度為4，那么它們之間就不能形成算子鏈。

(2) 下游節點的入度為1

下游節點的入度為1，即下游節點沒有來自其他節點的輸入。例如，如果一個算子有兩個輸入流，那么它就不能與上游算子形成算子鏈。

(3) 上下游節點都在同一個slot group中

上下游節點都必須在同一個slot group中，才能形成算子鏈。關于slot group的概念，將在后面的共享組部分詳細介紹。

(4) 下游節點的chain策略為ALWAYS

下游節點的chain策略為ALWAYS，表示可以與上下游鏈接，map、flatmap、filter 等默認是ALWAYS。

(5) 上游節點的chain策略為ALWAYS或HEAD

上游節點的chain策略為ALWAYS或HEAD，HEAD表示只能與下游鏈接，不能與上游鏈接，Source默認是HEAD。

(6) 兩個節點間數據分區方式是forward

兩個節點間的數據分區方式必須是forward，即數據不需要重新分區。

(7) 用戶沒有禁用chain

用戶沒有通過編程API禁用算子鏈。例如，沒有調用 disableChaining() 方法。

4. 算子鏈的控制方法

(1) 全局禁用算子鏈

可以通過調用 StreamExecutionEnvironment.disableOperatorChaining() 來全局禁用算子鏈。示例代碼如下：

StreamExecutionEnvironment env =StreamExecutionEnvironment.getExecutionEnvironment();
env.disableOperatorChaining();

(2) 從當前算子開始新鏈

可以通過在DataStream的operator后面調用 startNewChain() 來指示從該operator開始一個新的chain（與前面截斷，不會被chain到前面）。示例代碼如下：

DataStream<String> stream = env.fromElements("a","b","c");
stream.map(value -> value.toUpperCase())
.startNewChain()
.filter(value -> value.startsWith("A"));

(3) 禁用算子鏈

可以通過調用 disableChaining() 來指示該operator不參與chaining（不會與前后的operator chain一起）。示例代碼如下：

DataStream<String> stream = env.fromElements("a","b","c");
stream.map(value -> value.toUpperCase())
.disableChaining()
.filter(value -> value.startsWith("A"));

5. 算子鏈與并行度、Slot的最優搭配

(1) 并行度相同時的搭配

當上下游算子的并行度相同時，更容易形成算子鏈，從而減少線程切換和數據傳輸開銷。此時，應該盡量讓這些算子在同一個Slot中執行，以提高資源利用率。例如，如果Source算子和Map算子的并行度都為4，且滿足算子鏈的其他條件，那么它們可以形成算子鏈，只需要4個Slot來執行這兩個算子的任務。

(2) 并行度不同時的處理

如果上下游算子的并行度不同，無法形成算子鏈。這時需要根據具體情況調整并行度或分配Slot。例如，如果Source算子的并行度為2，而Map算子的并行度為4，可以考慮將Source算子的并行度提高到4，或者將Map算子的并行度降低到2，以嘗試形成算子鏈。如果無法調整并行度，就需要為每個算子的子任務分別分配Slot。

(3) 考慮算子的資源需求

對于資源密集型的算子，如aggregate、reduce、sum、window等，即使并行度相同，也可以考慮不與其他算子形成算子鏈，而是單獨分配Slot，以確保其有足夠的資源執行。例如，對于一個復雜的窗口操作，可以使用 startNewChain() 或 disableChaining() 方法將其與其他算子分開，使其獨享一個Slot的資源。而對于非資源密集型的算子，如source、map、sink等，可以盡量與其他算子形成算子鏈，共享Slot資源。

四、Flink共享組

1. 共享組的概念

在Apache Flink中，slotSharingGroup() 是一個用于控制算子（operator）之間資源共享的機制。它允許多個算子共享相同的slot（即資源容器）。Slot是Flink中的資源單位，slot共享可以提高資源利用率，但在某些情況下，我們希望更精細地控制不同算子的資源分配，slotSharingGroup 就提供了這種能力。

2. 共享組的作用

(1) 控制資源分配

將算子分配到不同的slot sharing group，可以將某些關鍵算子隔離出來，確保它們不會與其他算子爭用資源。例如，對于一些重要的窗口操作或聚合操作，可以為其分配獨立的slot sharing group，避免受到其他輕量級算子的干擾。

(2) 提高性能和穩定性

通過分組隔離，防止某些算子占用過多資源，從而影響其他算子的執行性能。例如，如果某個算子由于處理復雜度高或其他原因產生背壓，可能會影響同一slot sharing group中的其他算子。通過 slotSharingGroup() 隔離算子，可以減少背壓的擴散。

(3) 解決背壓問題

對于某些復雜的算子，可能會導致算子鏈中的其他算子受到背壓影響。通過將其分配到不同的slot sharing group，可以減少此類問題。例如，對于一個計算密集型的算子，可以將其分配到一個獨立的slot sharing group，避免對其他算子產生背壓。

3. 共享組的使用場景

(1) 算子資源隔離

當某些算子需要較高的資源或執行較復雜的邏輯時，可能希望將它們與其他輕量級算子隔離開來，避免干擾。比如某些窗口操作、聚合操作可能消耗大量內存和計算資源，此時可以為其分配獨立的slot sharing group。

(2) 優化并行度與資源利用率

在具有不同并行度的算子間，可以通過不同的slot sharing group來優化資源利用，避免算子在同一slot中因為并行度差異而出現負載不均的問題。例如，對于并行度較高的算子和并行度較低的算子，可以將它們分配到不同的slot sharing group，提高資源利用率。

(3) 避免背壓擴散

如果某個算子由于處理復雜度高或其他原因產生背壓，可能會影響同一slot sharing group中的其他算子。通過 slotSharingGroup() 隔離算子，可以減少背壓的擴散。例如，對于一個容易產生背壓的算子，可以將其分配到一個獨立的slot sharing group，避免影響其他算子的執行。

4. 共享組的代碼示例

// 定義兩個數據流
DataStream<String> stream1 = env.fromElements("a","b","c");
DataStream<String> stream2 = env.fromElements("1","2","3");
// 給第一個算子鏈設置 slotSharingGroup
stream1.map(value -> value.toUpperCase())
.slotSharingGroup("group1")
.filter(value -> value.startsWith("A"))
.slotSharingGroup("group1");
// 給第二個算子鏈設置不同的 slotSharingGroup
stream2.map(value -> value +"X")
.slotSharingGroup("group2")
.filter(value -> value.endsWith("X"))
.slotSharingGroup("group2");
// 匯聚兩個流并繼續處理
stream1.union(stream2)
.map(value ->"Processed: "+ value)
.slotSharingGroup("group3");
env.execute();

在上述代碼中，stream1 的算子被分配到了 "group1"，stream2 的算子被分配到了 "group2"，兩者之間的算子不會共享相同的slot，從而實現了資源隔離。最后，通過 union() 操作將兩個流合并并設置為 "group3"，合并后的流將使用一個新的共享組。

5. 共享組的效果

(1) 資源隔離

在上面的示例中，不同的算子鏈被分配到了不同的slot sharing group，實現了資源隔離。這可以確保關鍵算子不會受到其他算子的干擾，提高了作業的穩定性和性能。

(2) 優化資源分配

通過給不同的算子鏈分配不同的slot sharing group，Flink在作業執行時會為每個共享組分配不同的slot，避免了在同一個slot中同時運行可能會競爭資源的算子。這可以提高資源利用率，避免資源浪費。

(3) 減少資源爭用和背壓傳播

當某些復雜算子引發的背壓或資源消耗比較高時，其他不相關的算子不會受到其影響，從而提高了作業的穩定性和性能。例如，如果某個算子產生了背壓，由于它被分配到了一個獨立的slot sharing group，不會影響其他共享組中的算子。

6. 共享組的注意事項

(1) 默認設置

默認情況下，Flink的所有算子都屬于同一個默認的slot sharing group。如果不顯式設置 slotSharingGroup()，所有算子都會共享同一個slot。

(2) 資源不足問題

分配給一個slot sharing group的所有算子會被Flink盡可能分配到同一個slot中運行。如果算子的并行度較高，而集群資源不足，可能會導致部分算子不能有效共享slot，這時可以通過調整集群資源或者優化slot分配策略來解決。

(3) 鏈式操作影響

為可鏈式操作的算子設置不同的slot sharing group可能會導致鏈式操作 operator chains 產生割裂，從而改變性能。因此，在設置slot sharing group時，需要考慮算子之間的鏈式關系。

(4) 算子調度

slot 共享組僅僅意味著調度器可以使被分組的算子被部署到同一個slot中，但無法保證將被分組的算子部署在一起。如果被分組算子被部署到單獨的slot中，slot資源將從特定的資源組需求中派生而來。

7. 共享組與并行度、Slot、算子鏈的最優搭配

(1) 相同共享組內的搭配

將資源需求相似的算子分配到同一個共享組中，可以提高資源利用率。例如，將所有的非資源密集型算子（如source、map、sink）分配到一個共享組，將資源密集型算子（如aggregate、reduce、sum、window）分配到另一個共享組。在同一個共享組內，盡量讓并行度相同的算子形成算子鏈，在同一個Slot中執行。例如，在一個共享組內，Source算子和Map算子的并行度都為3，且滿足算子鏈的條件，它們可以形成算子鏈，只需要3個Slot。

(2) 不同共享組的隔離

對于資源需求差異較大或容易產生背壓的算子，分配到不同的共享組中，以避免資源競爭和背壓傳播。例如，將容易產生背壓的窗口操作算子分配到一個獨立的共享組，與其他算子隔離開來。這樣即使窗口操作算子產生背壓，也不會影響其他共享組中的算子。

(3) 綜合考慮并行度和Slot數量

在設置共享組時，需要綜合考慮整個作業的并行度和集群中可用的Slot數量。如果共享組內的算子并行度之和超過了可用的Slot數量，可能會導致部分任務等待或資源不足。因此，需要合理調整算子的并行度和共享組的分配，以確保作業能夠高效執行。