Understanding of Flink parallelism

Concept description

One Flink Program consists of multiple Operator form (source、transformation and sink).
One Operator By multiple parallel Task( Threads ) To execute , One Operator Parallel of Task( Threads ) The number is called the Operator( Mission ) Parallelism of (Parallel). That is, parallelism is relative to Operator Speaking of .

The official pair is listed below Operator Explanation ：

Operator

Node of a Logical Graph. An Operator performs a certain operation, which is usually executed by a Function. Sources and Sinks are special Operators for data ingestion and data egress.

Logical Graph

A logical graph is a directed graph where the nodes are Operators and the edges define input/output-relationships of the operators and correspond to data streams or data sets. A logical graph is created by submitting jobs from a Flink Application.

Logical graphs are also often referred to as dataflow graphs.

  The description in the source code ：

/** * Abstract base class for all operators. An operator is a source, sink, or it applies an operation * to one or more inputs, producing a result. * * @param <OUT> Output type of the records output by this operator */@Internalpublic abstract class Operator<OUT> implements Visitable<Operator<?>> {
   

/**     * Sets the parallelism for this contract instance. The parallelism denotes how many parallel     * instances of the user function will be spawned during the execution.     *     * @param parallelism The number of parallel instances to spawn. Set this value to {@link     *     ExecutionConfig#PARALLELISM_DEFAULT} to let the system decide on its own.     */    public void setParallelism(int parallelism) {        this.parallelism = parallelism;    }

Setting of parallelism

There can be 4 Levels to set Operator Parallelism of

1. Operator Level（ Operator level ）

2. Execution Environment Level（ Execution environment level ）

3. Client Level（ Client level ）

4. System Level（ System default level , Not recommended , Because it will affect all operations ）

1. Operator Level

Use the corresponding operator.setParallelism(xxx) that will do

2. Execution Environment Level

Use env.setParallelism(xxx) (env namely StreamExecutionEnvironment)

3. Client Level

Parallelism can be changed on the client job Submitted to the Flink Time setting .
about CLI client , Can pass -p Parameter specifies the degree of parallelism
./bin/flink run -p 3 ...

4. System Level

At the system level, you can set flink-conf.yaml In the document parallelism.default Property to specify the default parallelism for all execution environments .

4 Priority of setting methods

Priority of parallelism ： Operator level > env Level > Client level > System default level

That is to say, the value with high priority can override the value with low priority if it is set .

In addition, the parallelism set will not always be the same as that of the actual execution , such as If source Cannot be executed in parallel , Even if the parallelism is specified as multiple , It doesn't work ;kafka Read, etc .

In actual production , It is recommended to specify the degree of parallelism at the operator level , Convenient display and precise resource control .