Parallel Execution

This section describes how the parallel execution of programs can be configured in Flink. A Flinkprogram consists of multiple tasks (transformations/operators, data sources, and sinks). A task is split intoseveral parallel instances for execution and each parallel instance processes a subset of the task’sinput data. The number of parallel instances of a task is called its parallelism.

If you want to use savepoints you should also considersetting a maximum parallelism (or max parallelism). When restoring from a savepoint you canchange the parallelism of specific operators or the whole program and this setting specifiesan upper bound on the parallelism. This is required because Flink internally partitions stateinto key-groups and we cannot have +Inf number of key-groups because this would be detrimentalto performance.

Setting the Parallelism

The parallelism of a task can be specified in Flink on different levels:

Operator Level

The parallelism of an individual operator, data source, or data sink can be defined by calling itssetParallelism() method. For example, like this:

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");

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")

Execution Environment Level

As mentioned here Flinkprograms are executed in the context of an execution environment. Anexecution environment defines a default parallelism for all operators, data sources, and data sinksit executes. Execution environment parallelism can be overwritten by explicitly configuring theparallelism of an operator.

The default parallelism of an execution environment can be specified by calling thesetParallelism() method. To execute all operators, data sources, and data sinks with a parallelismof 3, set the default parallelism of the execution environment as follows:

final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.setParallelism(3);
DataStream<String> text = [...]
DataStream<Tuple2<String, Integer>> wordCounts = [...]
wordCounts.print();
env.execute("Word Count Example");

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")

Client Level

The parallelism can be set at the Client when submitting jobs to Flink. TheClient can either be a Java or a Scala program. One example of such a Client isFlink’s Command-line Interface (CLI).

For the CLI client, the parallelism parameter can be specified with -p. Forexample:

./bin/flink run -p 10 ../examples/*WordCount-java*.jar

In a Java/Scala program, the parallelism is set as follows:

try {
    PackagedProgram program = new PackagedProgram(file, args);
    InetSocketAddress jobManagerAddress = RemoteExecutor.getInetFromHostport("localhost:6123");
    Configuration config = new Configuration();
    Client client = new Client(jobManagerAddress, config, program.getUserCodeClassLoader());
    // set the parallelism to 10 here
    client.run(program, 10, true);
} catch (ProgramInvocationException e) {
    e.printStackTrace();
}

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())
    // set the parallelism to 10 here
    client.run(program, 10, true)
} catch {
    case e: Exception => e.printStackTrace
}

System Level

A system-wide default parallelism for all execution environments can be defined by setting theparallelism.default property in ./conf/flink-conf.yaml. See theConfiguration documentation for details.

Setting the Maximum Parallelism

The maximum parallelism can be set in places where you can also set a parallelism(except client level and system level). Instead of calling setParallelism() you callsetMaxParallelism() to set the maximum parallelism.

The default setting for the maximum parallelism is roughly operatorParallelism + (operatorParallelism / 2) witha lower bound of 128 and an upper bound of 32768.

Attention Setting the maximum parallelism to a very largevalue can be detrimental to performance because some state backends have to keep internal datastructures that scale with the number of key-groups (which are the internal implementation mechanism forrescalable state).