Parallel Execution

    If you want to use savepoints you should also consider setting a maximum parallelism (or ). 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 to performance.

    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:

    Java

    Scala

    1. val env = StreamExecutionEnvironment.getExecutionEnvironment
    3. val text = [...]
    4. val wordCounts = text
    5.     .flatMap{ _.split(" ") map { (_, 1) } }
    6.     .keyBy(_._1)
    7.     .window(TumblingEventTimeWindows.of(Time.seconds(5)))
    8. wordCounts.print()
    10. env.execute("Word Count Example")

    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 of 3, set the default parallelism of the execution environment as follows:

    Java

    Scala

    2. env.setParallelism(3)
    4. val text = [...]
    5. val wordCounts = text
    6.     .flatMap{ _.split(" ") map { (_, 1) } }
    7.     .keyBy(_._1)
    8.     .window(TumblingEventTimeWindows.of(Time.seconds(5)))
    9.     .sum(1)
    10. wordCounts.print()
    11. env.execute("Word Count Example")

    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 -p. For example:

    Java

    2. try {
    3.     PackagedProgram program = new PackagedProgram(file, args);
    5.     Configuration config = new Configuration();
    7.     Client client = new Client(jobManagerAddress, config, program.getUserCodeClassLoader());
    9.     // set the parallelism to 10 here
    10.     client.run(program, 10, true);
    12. } catch (ProgramInvocationException e) {
    13. }

    Scala

    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 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 call setMaxParallelism() to set the maximum parallelism.

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