在使用Apache Flink對數據進行處理時候，通常需要設置并行度。并行度是Apache Flink中一個非常重要的概念。設置合理的并行度能夠加快數據的處理效率，不合理的并行度會造成效率降低甚至是任務出錯。
Apache Flink程序包含多個任務(source,transformations/operators,sink)。這些任務使用幾個并行實例所進行執行，這些并行的實例稱之為并行度。

如何設置并行度

Apache Flink支持在不同的級別設置并行度。配置文件、env級別、算子級別。

1. 配置文件默認
   在我們提交一個Job的時候如果沒有考慮并行度的話，那么Flink會使用默認配置文件中的并行度。我們可以通過命令查看Flink配置文件的并行度。

$ cat flink-conf.yaml |grep "parallelism.default"
parallelism.default: 1

例如當前獲取到的并行度為1。也就是說當你不設置并行度的時候它就會使用配置文件默認的并行度 1。
2.  env級別
env的級別就是Environment級別。也就是通過Execution Environment來設置整體的Job并行度。

val env = Stream...
env.setParallelism(5)

3. 客戶端級別
   如果在執行Job時候，發現代碼中沒有設置并行度而又不修改配置文件的話，可以通過Client來設置Job的并行度。

./bin/flink run -p 5 ../wordCount-java*.jar

-p即設置WordCount的Job并行度為5。4.  算子級別
我們在編寫Flink項目時,可能對于不同的Operator設置不同的并行度，例如為了實現讀取Kafka的最高效讀取需要參考Kafka的partition的數量對并行度進行設置，在Sink時需要對于Sink的介質設置不同的并行度。這樣就會存在一個Job需要有多個并行度。這樣就需要用到算子級別的并行度設置

val env = Stream...
val text = ...
text.keyBy(XXX)
    .flatMap(XXX).setParallelism(5)  //計算時設置為5
    .addSink(XXXXX).setParallelism(1) //寫入數據庫時候設置為1

• 并行度的高級別會覆蓋低級別的配置。例如在算子中設置的策略會覆蓋配置文件中的parallelism。

從優先級上來看:  算子級別 > env級別 > Client級別 > 系統默認級別

在實際的使用中,我們需要設置合理的并行度來保證數據的高效處理，在一般情況下例如source，Sink等可能會需要不同的并行度來保證數據的快速讀取與寫入負載等。

 

并行度設置的數量

Apache Flink的并行度設置并不是說越大越好、數據處理的效率就越高。而是需要設置合理的并行度。那么何謂合理呢？
Apache Flink的 并行度取決于每個TaskManager上的slot數量而決定的。Flink的JobManager把任務分成子任務提交給slot進行執行。相同的slot共享相同的JVM資源，同時對Flink提供維護的心跳等信息。
slot是指TaskManagere的并發執行能力，通常來說TaskManager有多少核CPU也就會有多少個slot。這樣來看，我們設置的并行度其實是與TaskManager所有Slot數量有關的。