Flink的并发执行

本章描述如何在Flink中配置程序的并发执行，一个Flink程序可以由不同的task(如:transformations/opterators,data sources及data sinks等)组成，一个task会分发到多个并发实例中运行，并且每个并发实例处理task的部分输入数据集。一个task的并发实例数叫做parallelism。

如果你想使用savepoints的话，你需要设置一个最大并发数，当你从savepoints中重新获取数据时，你可以改变并发数，但是新的并发数必须大于先前的并发数。这样的要求是由于Flink内部将state根据key-group分到不同的partition中，我们不能添加key-group的数量，因为这对性能不友好。(这句不知道怎么翻译好，算了，不翻译了！)

设置Parallelism

task的parallelism可以在Flink的不同级别上指定:

算子(operator)级别

每个operator、data source或者data sink都可以通过调用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")

运行环境级别

正如此处所提醒的一样,Flink程序是在一个运行环境的上下文中运行的。一个运行环境为每个operator、data source和data sink的运行定义了一个默认的并发数。运行环境的并发数可以被每个算子确切的并发数配置所覆盖。

运行环境的默认并发数可以通过调用setParallelism()方法来指定。为了让所有的operator、data source和data sink以3个并发数来运行，你可按如下方法来设置运行环境的默认并发数:
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")

客户端级别

并发数可以在提交Job到Flink的客户端设置，客户端可以是Java或者Scala程序，典型的例子如:Flink的命令行接口(CLI).
对于CLI客户端，并发参数可以通过-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())

    // 将并发度设为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())

    // 将并发度设为10
    client.run(program, 10, true)

} catch {
    case e: Exception => e.printStackTrace
}

系统级别

影响所有运行环境的系统级别的默认并发度可以在./conf/flink-conf.yaml的parallelism.defaul项中指定，更多详情请参考这里

设置最大并发度

最大并发度可以在你设置并发度的地方设置(除开客户端级别和系统级别外)，你可以通过调用setMaxParallelism()方法来设置最大并发度。

默认的最大并发度大致为operatorParallelism + (parallelism/2) ，其中下限为127，上限为32768.

注意##