京東面試：如何合理設置 Flink 并行度？有哪些優化的點？
一、Flink并行度基礎概念
1. 什么是并行度
在Apache Flink中，并行度（Parallelism）是指一個Flink程序的并行執行能力。一個Flink程序由多個任務（task）組成，這些任務可以并行執行以提高處理效率。每個task包含多個并行執行的實例，且每一個實例都處理task輸入數據的一個子集。一個task的并行實例數被稱為該task的并行度。
Flink的并行架構由以下幾個關鍵組件組成：
- JobManager：協調分布式執行，如調度任務、協調檢查點等
- TaskManager：執行任務的工作節點，提供內存和處理能力
- Task Slot：TaskManager中的資源單位，每個slot可以執行一個并行任務實例
2. 并行度的重要性
合理設置并行度對Flink作業的性能至關重要，原因如下：
- 資源利用率：適當的并行度設置可以充分利用集群資源，避免資源浪費
- 處理吞吐量：更高的并行度通常意味著更高的數據處理吞吐量
- 延遲控制：合理的并行度可以減少數據處理的延遲
- 負載均衡：適當的并行度有助于在集群中均衡分配工作負載
- 成本效益：優化并行度可以在保證性能的同時降低資源成本
3. 影響并行度的因素
在設置Flink作業的并行度時，需要考慮以下因素：
- 數據量：處理的數據量越大，可能需要更高的并行度
- 計算復雜性：計算邏輯越復雜，可能需要更高的并行度
- 可用資源：集群的可用資源（CPU、內存等）限制了最大可能的并行度
- 數據傾斜：數據分布不均勻可能導致某些并行實例負載過重
- 狀態大小：有狀態操作的狀態大小會影響內存使用和并行度選擇
- 網絡傳輸：過高的并行度可能導致過多的網絡傳輸開銷
二、并行度配置級別與方法
Flink提供了多個級別的并行度配置，從最具體到最一般依次為：
1. 算子級別（最高優先級）
可以為單個算子（operator）設置特定的并行度，這將覆蓋所有其他級別的設置：
// Java示例
DataStream<String> dataStream = env.fromElements("a","b","c");
// 為map算子設置并行度為2
dataStream.map(s -> s.toUpperCase()).setParallelism(2);
// 為keyBy/sum算子設置并行度為3
dataStream.keyBy(value -> value).sum(0).setParallelism(3);
# Python示例
data_stream = env.from_elements("a", "b", "c")
# 為map算子設置并行度為2
data_stream.map(lambda s: s.upper()).set_parallelism(2)
# 為keyBy/sum算子設置并行度為3
data_stream.key_by(lambda x: x).sum(0).set_parallelism(3)2. 執行環境級別
可以在StreamExecutionEnvironment中設置所有算子的默認并行度：
// Java示例
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.setParallelism(4);// 設置默認并行度為4
// 此處所有算子將使用并行度4，除非單獨指定
DataStream<String> dataStream = env.fromElements("a","b","c");
dataStream.map(s -> s.toUpperCase());
dataStream.keyBy(value -> value).sum(0);
# Python示例
env = StreamExecutionEnvironment.get_execution_environment()
env.set_parallelism(4) # 設置默認并行度為4
# 此處所有算子將使用并行度4，除非單獨指定
data_stream = env.from_elements("a", "b", "c")
data_stream.map(lambda s: s.upper())
data_stream.key_by(lambda x: x).sum(0)3. 客戶端級別
在提交作業時，可以通過命令行參數指定并行度：
# 使用命令行參數設置并行度
bin/flink run -p 8 examples/streaming/WordCount.jar4. 系統級別（最低優先級）
可以在Flink配置文件（flink-conf.yaml）中設置集群范圍的默認并行度：
# 在flink-conf.yaml中設置
parallelism.default:25. 并行度配置優先級
當多個級別同時設置并行度時，優先級從高到低為：
- 算子級別 (setParallelism())
- 執行環境級別(env.setParallelism()) 3
- 客戶端級別 (命令行 -p 參數) 4. 系統級別 (flink-conf.yaml)
三、自適應并行度與自動優化
1. 自適應批處理調度器
Flink引入了AdaptiveBatchScheduler調度器，該調度器能夠自動調整批處理作業的并行度，無需手動設置。它根據輸入數據量和可用資源自動推導出最優的并行度配置。
(1) 自動推導算子并行度
AdaptiveBatchScheduler支持自動推導算子并行度，主要優勢包括：
- 推作業用戶可以從并行度調優中解放出來
- 根據數據量自動推導并行度可以更好地適應數據變化
- SQL作業的算子也可以分配不同的并行度
(2) 啟用自動并行度推導
要使用AdaptiveBatchScheduler自動推導算子并行度，需要：
① 啟用自動并行度推導：
// Java示例
Configuration configuration=newConfiguration();
// 啟用自適應批處理調度器的自動并行度功能
configuration.set(ExecutionOptions.BATCH_ADAPTIVE_AUTO_PARALLELISM_ENABLED,true);
// 設置自動并行度的最小值
configuration.set(ExecutionOptions.BATCH_ADAPTIVE_AUTO_PARALLELISM_MIN_PARALLELISM,1);
// 設置自動并行度的最大值
configuration.set(ExecutionOptions.BATCH_ADAPTIVE_AUTO_PARALLELISM_MAX_PARALLELISM,64);
// 設置每個任務平均處理的數據量
configuration.set(ExecutionOptions.BATCH_ADAPTIVE_AUTO_PARALLELISM_AVG_DATA_VOLUME_PER_TASK, MemorySize.ofMebiBytes(8));
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(configuration);② 也可以通過配置文件（flink-conf.yaml）啟用：
execution.batch.adaptive.auto-parallelism.enabled:true
execution.batch.adaptive.auto-parallelism.min-parallelism:1
execution.batch.adaptive.auto-parallelism.max-parallelism:64
execution.batch.adaptive.auto-parallelism.avg-data-volume-per-task: 8mb2. 自適應數據分發
AdaptiveBatchScheduler還支持其他優化功能：
(1) 自適應Broadcast Join
對于廣播連接（Broadcast Join），調度器可以根據數據量自動選擇最佳的廣播策略，減少不必要的數據傳輸。
(2) 自適應Skewed Join優化
對于數據傾斜的連接操作，調度器可以自動檢測并優化數據分布不均勻的情況，提高連接操作的性能。
四、性能調優策略
1. 資源配置優化
(1) TaskManager和Slot配置
TaskManager和Slot的合理配置對并行度優化至關重要：
- TaskManager數量：通常與集群物理節點數相關
- 每個TaskManager的Slot數量：通常設置為每個TaskManager的CPU核心數
- 內存配置：需要根據作業特性合理分配TaskManager的內存
# TaskManager配置示例
taskmanager.numberOfTaskSlots:8
taskmanager.memory.process.size: 4096m(2) 資源組（Resource Group）
資源組允許將相關的算子分組，以便它們在同一個TaskManager上執行，減少網絡傳輸：
// Java示例
// 定義資源組
ResourceSpec spec = ResourceSpec.newBuilder()
.setCpuCores(1.0)
.setTaskHeapMemoryMB(512)
.build();
// 將算子分配到資源組
dataStream.map(newMyMapper()).slotSharingGroup("group1").setResources(spec);2. 算子鏈（Operator Chaining）
算子鏈是Flink的一項重要優化，它將多個算子合并到一個任務中執行，減少了任務間的數據傳輸開銷：
(1) 啟用/禁用算子鏈
// Java示例
// 全局禁用算子鏈
env.disableOperatorChaining();
// 為特定算子禁用鏈接
dataStream.map(newMyMapper()).disableChaining();
// 開始新的鏈
dataStream.map(newMyMapper()).startNewChain();(2) 算子鏈最佳實踐
- 將計算密集型算子與IO密集型算子分開鏈接
- 避免將狀態較大的算子鏈接在一起
- 考慮將具有相似資源需求的算子鏈接在一起
3. 數據傾斜處理
數據傾斜是影響并行度效率的主要因素之一：
(1) 識別數據傾斜
- 使用Flink Web UI監控任務執行
- 觀察各個子任務的處理記錄數和處理時間
- 檢查背壓（backpressure）指標
(2) 解決數據傾斜的策略
① 預聚合：在keyBy之前進行局部聚合，減少數據量
// Java示例 - 兩階段聚合處理數據傾斜
dataStream
.map(newPreAggregateFunction()) // 第一階段：局部預聚合
.keyBy(value -> value.getKey())
.window(TumblingEventTimeWindows.of(Time.seconds(10)))
.reduce(newAggregateFunction()) // 第二階段：全局聚合② Key重分區：為熱點key添加隨機前綴，將一個熱點key分散到多個任務
// Java示例 - 使用隨機前綴重分區熱點key
dataStream
.map(event ->{
// 為熱點key添加隨機前綴
if(isHotKey(event.getKey())){
int randomPrefix = ThreadLocalRandom.current().nextInt(parallelism);
returnnew Tuple2<>(randomPrefix +"_"+event.getKey(), event.getValue());
}else{
returnnew Tuple2<>(event.getKey(), event.getValue());
}
})
.keyBy(tuple -> tuple.f0) // 使用新key進行分區
.window(TumblingEventTimeWindows.of(Time.seconds(10)))
.reduce(newAggregateFunction())
// 最后移除前綴
.map(tuple ->new Tuple2<>(removePrefix(tuple.f0), tuple.f1));4. 狀態管理優化
對于有狀態的操作，狀態管理對并行度優化也很重要：
(1) 狀態后端選擇
Flink提供了三種狀態后端，根據作業特性選擇合適的狀態后端：
- MemoryStateBackend：小狀態，低延遲，不需要恢復
- FsStateBackend：大狀態，低延遲，可靠恢復
- RocksDBStateBackend：超大狀態，較高延遲，可增量檢查點
// Java示例 - 配置狀態后端
// 內存狀態后端
env.setStateBackend(newMemoryStateBackend());
// 文件系統狀態后端
env.setStateBackend(newFsStateBackend("hdfs://namenode:9000/flink/checkpoints"));
// RocksDB狀態后端
env.setStateBackend(newRocksDBStateBackend("hdfs://namenode:9000/flink/checkpoints",true));(2) 狀態大小與并行度的關系
- 增加并行度會將狀態分散到更多的任務實例中
- 過大的狀態可能導致內存壓力，影響性能
- 考慮使用RocksDBStateBackend處理超大狀態
