Flink的并發(fā)執(zhí)行
本章描述如何在Flink中配置程序的并發(fā)執(zhí)行癌椿，一個Flink程序可以由不同的task(如:transformations/opterators,data sources及data sinks等)組成，一個task會分發(fā)到多個并發(fā)實例中運行，并且每個并發(fā)實例處理task的部分輸入數(shù)據(jù)集贷岸。一個task的并發(fā)實例數(shù)叫做parallelism篮灼。
如果你想使用savepoints的話作烟，你需要設(shè)置一個最大并發(fā)數(shù)河闰，當(dāng)你從savepoints中重新獲取數(shù)據(jù)時珍促，你可以改變并發(fā)數(shù)响委，但是新的并發(fā)數(shù)必須大于先前的并發(fā)數(shù)新思。這樣的要求是由于Flink內(nèi)部將state根據(jù)key-group分到不同的partition中，我們不能添加key-group的數(shù)量赘风，因為這對性能不友好夹囚。(這句不知道怎么翻譯好，算了邀窃，不翻譯了荸哟！)
設(shè)置Parallelism
task的parallelism可以在Flink的不同級別上指定:
算子(operator)級別
每個operator、data source或者data sink都可以通過調(diào)用setParallelism()方法來指定parallelism蛔翅，例如:
Java代碼:
final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
DataStream<String> text = [...]
DataStream<Tuple2<String, Integer>> wordCounts = text
.flatMap(new LineSplitter())
.keyBy(0)
.timeWindow(Time.seconds(5))
.sum(1).setParallelism(5);
wordCounts.print();
env.execute("Word Count Example");
Scala代碼:
val env = StreamExecutionEnvironment.getExecutionEnvironment
val text = [...]
val wordCounts = text
.flatMap{ _.split(" ") map { (_, 1) } }
.keyBy(0)
.timeWindow(Time.seconds(5))
.sum(1).setParallelism(5)
wordCounts.print()
env.execute("Word Count Example")
運行環(huán)境級別
正如此處所提醒的一樣,Flink程序是在一個運行環(huán)境的上下文中運行的敲茄。一個運行環(huán)境為每個operator、data source和data sink的運行定義了一個默認(rèn)的并發(fā)數(shù)山析。運行環(huán)境的并發(fā)數(shù)可以被每個算子確切的并發(fā)數(shù)配置所覆蓋堰燎。
運行環(huán)境的默認(rèn)并發(fā)數(shù)可以通過調(diào)用setParallelism()方法來指定。為了讓所有的operator笋轨、data source和data sink以3個并發(fā)數(shù)來運行秆剪，你可按如下方法來設(shè)置運行環(huán)境的默認(rèn)并發(fā)數(shù):
Java代碼:
final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.setParallelism(3);
DataStream<String> text = [...]
DataStream<Tuple2<String, Integer>> wordCounts = [...]
wordCounts.print();
env.execute("Word Count Example");
Scala代碼:
val env = StreamExecutionEnvironment.getExecutionEnvironment
env.setParallelism(3)
val text = [...]
val wordCounts = text
.flatMap{ _.split(" ") map { (_, 1) } }
.keyBy(0)
.timeWindow(Time.seconds(5))
.sum(1)
wordCounts.print()
env.execute("Word Count Example")
客戶端級別
并發(fā)數(shù)可以在提交Job到Flink的客戶端設(shè)置，客戶端可以是Java或者Scala程序爵政，典型的例子如:Flink的命令行接口(CLI).
對于CLI客戶端仅讽，并發(fā)參數(shù)可以通過-p來指定，例如:
./bin/flink run -p 10 ../examples/*WordCount-java*.jar
在Java/Scala程序中钾挟，可以按如下方式指定:
Java代碼:
try {
PackagedProgram program = new PackagedProgram(file, args)
InetSocketAddress jobManagerAddress = RemoteExecutor.getInetFromHostport("localhost:6123")
Configuration config = new Configuration()
Client client = new Client(jobManagerAddress, new Configuration(), program.getUserCodeClassLoader())
// 將并發(fā)度設(shè)為10
client.run(program, 10, true)
} catch {
case e: Exception => e.printStackTrace
}
Scala代碼:
try {
val program = new PackagedProgram(file, args)
val jobManagerAddress = RemoteExecutor.getInetFromHostport("localhost:6123")
val config = new Configuration()
val client = new Client(jobManagerAddress, config, program.getUserCodeClassLoader())
// 將并發(fā)度設(shè)為10
client.run(program, 10, true)
} catch {
case e: Exception => e.printStackTrace
}
系統(tǒng)級別
影響所有運行環(huán)境的系統(tǒng)級別的默認(rèn)并發(fā)度可以在./conf/flink-conf.yaml的parallelism.defaul項中指定洁灵，更多詳情請參考這里
設(shè)置最大并發(fā)度
最大并發(fā)度可以在你設(shè)置并發(fā)度的地方設(shè)置(除開客戶端級別和系統(tǒng)級別外)，你可以通過調(diào)用setMaxParallelism()方法來設(shè)置最大并發(fā)度掺出。
默認(rèn)的最大并發(fā)度大致為operatorParallelism + (parallelism/2) 徽千，其中下限為127，上限為32768.
