设置并行度

Flink应用程序在一个像集群这样的分布式环境中并行执行。当一个数据流程序提交到作业管理器执行时，系统将会创建一个数据流图，然后准备执行需要的操作符。每一个操作符将会并行化到一个或者多个任务中去。每个算子的并行任务都会处理这个算子的输入流中的一份子集。一个算子并行任务的个数叫做算子的并行度。它决定了算子执行的并行化程度，以及这个算子能处理多少数据量。

算子的并行度可以在执行环境这个层级来控制，也可以针对每个不同的算子设置不同的并行度。默认情况下，应用程序中所有算子的并行度都将设置为执行环境的并行度。执行环境的并行度（也就是所有算子的默认并行度）将在程序开始运行时自动初始化。如果应用程序在本地执行环境中运行，并行度将被设置为CPU的核数。当我们把应用程序提交到一个处于运行中的Flink集群时，执行环境的并行度将被设置为集群默认的并行度，除非我们在客户端提交应用程序时显式的设置好并行度。

通常情况下，将算子的并行度定义为和执行环境并行度相关的数值会是个好主意。这允许我们通过在客户端调整应用程序的并行度就可以将程序水平扩展了。我们可以使用以下代码来访问执行环境的默认并行度。

我们还可以重写执行环境的默认并行度，但这样的话我们将再也不能通过客户端来控制应用程序的并行度了。

算子默认的并行度也可以通过重写来明确指定。在下面的例子里面，数据源的操作符将会按照环境默认的并行度来并行执行，map操作符的并行度将会是默认并行度的2倍，sink操作符的并行度为2。

StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment;
int defaultP = env.getParallelism;
env
  .addSource(new CustomSource)
  .map(new MyMapper)
  .setParallelism(defaultP * 2)
  .print()
  .setParallelism(2);

当我们通过客户端将应用程序的并行度设置为16并提交执行时，source操作符的并行度为16，mapper并行度为32，sink并行度为2。如果我们在本地环境运行应用程序的话，例如在IDE中运行，机器是8核，那么source任务将会并行执行在8个任务上面，mapper运行在16个任务上面，sink运行在2个任务上面。

并行度是动态概念，任务槽数量是静态概念。并行度<=任务槽数量。一个任务槽最多运行一个并行度。