一、Flink 架構

Flink 架構 | Apache Flink

二、設置TaskManager、Slot和Parallelism

在Apache Flink中，設置TaskManager、Slot和Parallelism是配置Flink集群性能和資源利用的關鍵步驟。以下是關于如何設置這些參數的詳細指南：

1. TaskManager 設置

TaskManager是Flink集群中負責執行作業的節點。關于TaskManager的設置，主要關注其數量和資源分配。

1. TaskManager數量：根據集群規模和作業需求確定TaskManager的數量。例如，如果集群資源充足且作業并發度高，可以增加TaskManager的數量以提高處理能力。
2. 資源分配：為每個TaskManager分配適當的內存和CPU資源。這取決于集群的硬件配置和作業的資源需求。確保為TaskManager分配足夠的資源以確保作業可以高效運行。

2. Slot 設置

Slot是TaskManager上用于執行作業的資源單元。一個Slot可以并行運行一個作業的子任務。

1. Slot數量：每個TaskManager上的Slot數量決定了該TaskManager可以并行運行的作業子任務數。Slot數量通常根據TaskManager的內存和CPU資源來確定。例如，如果TaskManager有2GB內存和1個CPU核心，并且每個Slot需要1GB內存和0.5個CPU核心，則該TaskManager可以設置2個Slot。
2. 資源分配：每個Slot會分配到一定的內存和CPU資源。這些資源應該根據作業的需求和TaskManager的總資源進行合理分配。

3. Parallelism 設置

Parallelism決定了Flink作業的并行度，即作業可以并行執行的程度。

1. 默認并行度：在Flink配置文件中，可以指定默認并行度（parallelism.default）。如果作業沒有指定并行度，則使用默認并行度。
2. 作業級并行度：在提交作業時，可以通過命令行參數（-p）或編程API（env.setParallelism()）為整個作業設置并行度。這將作為作業的默認并行度，但可以被單個算子的并行度設置覆蓋。
3. 算子級并行度：在Flink程序中，可以為每個算子單獨設置并行度。這可以通過在算子鏈的末尾調用setParallelism()方法來實現。算子級并行度的優先級高于作業級并行度和默認并行度。

4. 總結

• 設置TaskManager的數量和資源分配以適應集群規模和作業需求。
• 根據TaskManager的資源為每個TaskManager設置適當的Slot數量。
• 根據作業的需求和集群的資源設置作業的默認并行度、作業級并行度和算子級并行度。

?5. 阿里云 實時計算Flink版 參數示例?

三、Flink SQL性能調優與配置

在使用Flink SQL進行數據處理時，性能調優是確保系統高效運行的關鍵。以下是一些常見的調優配置和策略，它們可以幫助您優化Flink SQL作業的性能。

1. 微批處理（Mini-Batch）

Flink SQL支持微批處理，通過組合多個小批次來減少任務調度的開銷。當啟用微批處理時，Flink會嘗試將多個小批次合并成一個較大的批次進行處理。

# 啟用微批處理

table.exec.mini-batch.enabled: 'true'

# 設置允許的最大延遲時間，超過該時間將不再等待更多數據而直接發送當前批次

table.exec.mini-batch.allow-latency: 2s

2. 算子鏈優化（Operator Chaining）

算子鏈優化是一種減少任務間數據傳輸開銷的策略。通過將多個算子鏈接在一起，可以減少序列化和反序列化的開銷，并提高數據傳輸的效率。

# 默認情況下，Flink會嘗試自動進行算子鏈優化

# 如果需要禁用此功能，可以設置為false

pipeline.operator-chaining: 'false'

注意：通常建議保持算子鏈優化開啟（'true'），以獲得更好的性能。

3. Hash Shuffle

在Flink中，Keyed Streams使用hash shuffle策略將數據分發到下游的并行任務。這有助于確保具有相同key的數據被發送到同一個下游任務，從而進行高效的聚合或連接操作。

對于Flink SQL中的sink，如果其接受的是Keyed Stream，并且需要確保數據的順序性，可以使用FORCE關鍵字來強制使用hash shuffle。

# 強制使用hash shuffle

table.exec.sink.keyed-shuffle: FORCE

注意：在Flink SQL中，您通常不需要手動配置這個參數，因為Flink會根據作業的特性和需求自動選擇合適的shuffle策略。

4. Hash Join

Hash Join是一種基于哈希表的連接算法，適用于等值連接場景。它通過將一個表的數據加載到哈希表中，然后掃描另一個表并與哈希表中的數據進行比較來實現連接。

在Flink SQL中，可以使用Hint（提示）來建議優化器使用Hash Join。但是，請注意，這只是一個建議，優化器可能會根據實際情況選擇其他連接策略。

SELECT /*+ SHUFFLE_HASH(t1,t2) */ * FROM table1 t1 JOIN table2 t2 ON t1.key = t2.key

5. 設置Sink的并行度

Sink的并行度決定了數據寫入外部系統時的并行度。可以根據外部系統的性能和Flink作業的需求來設置合適的并行度。

在Flink SQL中，可以通過DDL語句或API來設置Sink的并行度。以下是一個示例DDL語句：

CREATE TABLE sink_table (

... -- 定義表結構

) WITH (

... -- 其他配置選項

'sink.parallelism' = '4' -- 設置并行度為4

);

或者，在Flink作業提交時通過API來動態設置Sink的并行度。