说明

可以在以下设置并行度：

1. 算子级别
2. 运行环境
3. 客户端CLI（启动命令参数）
4. 系统配置文件

提示：

• 算子级别 > 运行环境 > 客户端CLI > 系统配置文件
• 优先级高的，可以覆盖优先级低的

算子级别

每个 operator、data source 或者 data sink 都可以通过调用 setParallelism() 方法来指定 parallelism

如下：

dataStream
    .flatMap(...)
    .setParallelism(5)
    .keyBy()

不是所有的算子都能设置并行度

flatMap() 可以设置
keyBy() 不可以设置，并行度受全局控制

dataStream
    .flatMap(...)
    .setParallelism(5) // flatMap 并行度是5
    .keyBy() // 受全局控制，不与 flatMap 并行度相同

运行环境

Flink程序是在一个运行环境的上下文中运行的。运行环境为每个operator、data source 和 data sink 的运行定义了一个 默认的并发数。

运行环境的并发数可以被每个算子的并发数所覆盖。

运行环境的默认并发数可以通过调用 setParallelism() 方法来指定：

StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.setParallelism(3);

客户端CLI（启动命令参数）

并发数可以在提交Job到Flink的客户端设置，并发参数可以通过 -p 来指定，例如:

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

系统配置文件

系统级别的默认并发度在下面文件中配置：

./conf/flink-conf.yaml

的

parallelism.default: 1

最大并发度

最大并发度可以在设置并发度的地方设置(除开客户端级别和系统级别外)

可以通过调用 setMaxParallelism() 方法来设置最大并发度，如下：

运行环境设置：

env.setMaxParallelism(8)

算子设置（部分设置）：

print().setParallelism(2)

上限、下线

默认的最大并发度大致为 operatorParallelism + (parallelism/2)

• 其中下限为 127
• 上限为 32768

注意:

• setParallelism() 必须小于 setMaxParallelism()设置的最大并行度。

• 某些算子无法设置并行度，如：

  • socketTextStream
  • 本地模式并行度默认为 cpu核数

• 并行度改变会影响影响task数量，如果taskslots数量不满足要求，会导致任务没有足够的资源分配

参考：

https://www.jianshu.com/p/9c301ff1a9b2
https://blog.csdn.net/weixin_38251332/article/details/124179583

原文出处：/show_1IX44XYOJPop.html