This section describes how the parallel execution of programs can be configured in Flink. A Flink program consists of multiple tasks (transformations/operators, data sources, and sinks). A task is split into several parallel instances for execution and each parallel instance processes a subset of the task’s input data. The number of parallel instances of a task is called its parallelism.
If you want to use savepoints you should also consider
setting a maximum parallelism (or
max parallelism). When restoring from a savepoint you can
change the parallelism of specific operators or the whole program and this setting specifies
an upper bound on the parallelism. This is required because Flink internally partitions state
into key-groups and we cannot have
+Inf number of key-groups because this would be detrimental
The parallelism of a task can be specified in Flink on different levels:
The parallelism of an individual operator, data source, or data sink can be defined by calling its
setParallelism() method. For example, like this:
As mentioned here Flink programs are executed in the context of an execution environment. An execution environment defines a default parallelism for all operators, data sources, and data sinks it executes. Execution environment parallelism can be overwritten by explicitly configuring the parallelism of an operator.
The default parallelism of an execution environment can be specified by calling the
setParallelism() method. To execute all operators, data sources, and data sinks with a parallelism
3, set the default parallelism of the execution environment as follows:
The parallelism can be set at the Client when submitting jobs to Flink. The Client can either be a Java or a Scala program. One example of such a Client is Flink’s Command-line Interface (CLI).
For the CLI client, the parallelism parameter can be specified with
./bin/flink run -p 10 ../examples/*WordCount-java*.jar
In a Java/Scala program, the parallelism is set as follows:
A system-wide default parallelism for all execution environments can be defined by setting the
parallelism.default property in
./conf/flink-conf.yaml. See the
Configuration documentation for details.
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 call
setMaxParallelism() to set the maximum parallelism.
The default setting for the maximum parallelism is roughly
operatorParallelism + (operatorParallelism / 2) with
a lower bound of
128 and an upper bound of
Attention Setting the maximum parallelism to a very large value can be detrimental to performance because some state backends have to keep internal data structures that scale with the number of key-groups (which are the internal implementation mechanism for rescalable state).