Flink parallelism/maxParallelism 作用源码

亦凉 2021-09-25 03:26 361阅读 0赞

# Parallel Execution #

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 to performance.

本节描述如何在Flink中配置程序的并行执行。Flink程序由多个任务组成（转换/运算符、数据源和接收器）。一个任务被分成几个并行实例来执行，每个并行实例处理任务输入数据的一个子集。任务的并行实例数称为其并行性。

**如果要使用保存点，还应考虑设置最大并行度（或最大并行度）。从保存点还原时，可以更改特定运算符或整个程序的并行度，此设置指定并行度的上限。这是必需的，因为Flink在内部将状态划分为键组，并且我们不能有无限个键组，因为这将对性能有害。**

## 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 its `setParallelism()` method. For example, like this:

*  [**Java**][Java]
 *  [**Scala**][Scala]

final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
 
 DataStream<String> text = [...]
 DataStream<Tuple2<String, Integer>> wordCounts = text
 .flatMap(new LineSplitter())
 .keyBy(value -> value.f0)
 .window(TumblingEventTimeWindows.of(Time.seconds(5)))
 .sum(1).setParallelism(5);
 
 wordCounts.print();
 
 env.execute("Word Count Example");

### Execution Environment Level ###

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

*  [**Java**][Java 1]
 *  [**Scala**][Scala 1]

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

### Client Level ###

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:

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

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

*  [**Java**][Java 2]
 *  [**Scala**][Scala 2]

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();
    }

### System Level ###

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.

## 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 `128` and an upper bound of `32768`.

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

**设置最大平行度**

**最大并行度可以在您还可以设置并行度的地方设置（客户机级别和系统级别除外）。不是调用setParallelism（），而是调用setMaxParallelism（）来设置最大并行度。**

**最大并行度的默认设置大致为operatorParallelism+（operatorParallelism/2），下限为128，上限为32768。**

**注意：将最大并行度设置为非常大的值可能会对性能造成不利影响，因为某些状态后端必须保持内部数据结构随密钥组的数量而扩展（这是可重缩放状态的内部实现机制）。**

实例：

![20210125101614750.png][]

[https://ci.apache.org/projects/flink/flink-docs-stable/dev/parallel.html][https_ci.apache.org_projects_flink_flink-docs-stable_dev_parallel.html]

源码：

SingleOutputStreamOperator.java

/**
 	 * Sets an ID for this operator.
 	 *
 	 * The specified ID is used to assign the same operator ID across job
 	 * submissions (for example when starting a job from a savepoint).
 	 *
 	 * Important: this ID needs to be unique per
 	 * transformation and job. Otherwise, job submission will fail.
 	 *
 	 * @param uid The unique user-specified ID of this transformation.
 	 * @return The operator with the specified ID.
 	 */
 	@PublicEvolving
 	public SingleOutputStreamOperator<T> uid(String uid) {
 		transformation.setUid(uid);
 		return this;
 	}
 
 	
 
 	/**
 	 * Sets the parallelism for this operator.
 	 *
 	 * @param parallelism
 	 * The parallelism for this operator.
 	 * @return The operator with set parallelism.
 	 */
 	public SingleOutputStreamOperator<T> setParallelism(int parallelism) {
 		Preconditions.checkArgument(canBeParallel() || parallelism == 1,
 				"The parallelism of non parallel operator must be 1.");
 
 		transformation.setParallelism(parallelism);
 
 		return this;
 	}
 
 	/**
 	 * Sets the maximum parallelism of this operator.
 	 *
 	 * The maximum parallelism specifies the upper bound for dynamic scaling. It also defines the
 	 * number of key groups used for partitioned state.
 	 *
 	 * @param maxParallelism Maximum parallelism
 	 * @return The operator with set maximum parallelism
 	 */
 	@PublicEvolving
 	public SingleOutputStreamOperator<T> setMaxParallelism(int maxParallelism) {
 		Preconditions.checkArgument(maxParallelism > 0,
 				"The maximum parallelism must be greater than 0.");
 
 		Preconditions.checkArgument(canBeParallel() || maxParallelism == 1,
 				"The maximum parallelism of non parallel operator must be 1.");
 
 		transformation.setMaxParallelism(maxParallelism);
 
 		return this;
 	}

Transformation.java

@Internal
 public abstract class Transformation<T> {
 
 	// Has to be equal to StreamGraphGenerator.UPPER_BOUND_MAX_PARALLELISM
 	public static final int UPPER_BOUND_MAX_PARALLELISM = 1 << 15;
 
 
 /**
 	 * The flag value indicating use of the default parallelism. This value can
 	 * be used to reset the parallelism back to the default state.
 	 */
 	public static final int PARALLELISM_DEFAULT = -1;
 
 
 /**
 	 * Returns the parallelism of this {@code Transformation}.
 	 */
 	public int getParallelism() {
 		return parallelism;
 	}
 
 	/**
 	 * Sets the parallelism of this {@code Transformation}.
 	 * 大于0 或 == -1 是符合条件的
 	 * @param parallelism The new parallelism to set on this {@code Transformation}.
 	 */
 	public void setParallelism(int parallelism) {
 		Preconditions.checkArgument(
 				parallelism > 0 || parallelism == ExecutionConfig.PARALLELISM_DEFAULT,
 				"The parallelism must be at least one, or ExecutionConfig.PARALLELISM_DEFAULT (use system default).");
 		this.parallelism = parallelism;
 	}
 
 	/**
 	 * Gets the maximum parallelism for this stream transformation.
 	 *
 	 * @return Maximum parallelism of this transformation.
 	 */
 	public int getMaxParallelism() {
 		return maxParallelism;
 	}
 
 	/**
 * 大于0 且<= 2^15 是符合条件的
 	 * Sets the maximum parallelism for this stream transformation.
 	 *
 	 * @param maxParallelism Maximum parallelism for this stream transformation.
 	 */
 	public void setMaxParallelism(int maxParallelism) {
 		Preconditions.checkArgument(maxParallelism > 0
 						&& maxParallelism <= UPPER_BOUND_MAX_PARALLELISM,
 				"Maximum parallelism must be between 1 and " + UPPER_BOUND_MAX_PARALLELISM
 						+ ". Found: " + maxParallelism);
 		this.maxParallelism = maxParallelism;
 	}
 
 }

Preconditions.java

/**
    	 * Checks the given boolean condition, and throws an {@code IllegalArgumentException} if
    	 * the condition is not met (evaluates to {@code false}). The exception will have the
    	 * given error message.
    	 *
    	 * @param condition The condition to check
    	 * @param errorMessage The message for the {@code IllegalArgumentException} that is thrown if the check fails.
    	 *
    	 * @throws IllegalArgumentException Thrown, if the condition is violated.
    	 */
    	public static void checkArgument(boolean condition, @Nullable Object errorMessage) {
    		if (!condition) {
    			throw new IllegalArgumentException(String.valueOf(errorMessage));
    		}
    	}

flink 1.9.2

##  ##

[savepoints]: https://ci.apache.org/projects/flink/flink-docs-release-1.12/ops/state/savepoints.html
[Java]: https://ci.apache.org/projects/flink/flink-docs-stable/dev/parallel.html#tab_Java_0
[Scala]: https://ci.apache.org/projects/flink/flink-docs-stable/dev/parallel.html#tab_Scala_0
[here]: https://ci.apache.org/projects/flink/flink-docs-release-1.12/dev/datastream_api.html#anatomy-of-a-flink-program
[Java 1]: https://ci.apache.org/projects/flink/flink-docs-stable/dev/parallel.html#tab_Java_1
[Scala 1]: https://ci.apache.org/projects/flink/flink-docs-stable/dev/parallel.html#tab_Scala_1
[Java 2]: https://ci.apache.org/projects/flink/flink-docs-stable/dev/parallel.html#tab_Java_2
[Scala 2]: https://ci.apache.org/projects/flink/flink-docs-stable/dev/parallel.html#tab_Scala_2
[Configuration]: https://ci.apache.org/projects/flink/flink-docs-release-1.12/deployment/config.html
[20210125101614750.png]: /images/20210923/9b8d33efbbb641628f2d1927ad324710.png
[https_ci.apache.org_projects_flink_flink-docs-stable_dev_parallel.html]: https://ci.apache.org/projects/flink/flink-docs-stable/dev/parallel.html