4.3 一些重要概念

我们现在已经了解Flink运行时的核心组件和整体架构，也明白了不同场景下作业提交的具体流程，但有些细节还需要进一步思考：一个具体的作业，是怎样从我们编写的代码，转换成TaskManager可以执行的任务的呢？JobManager收到提交的作业，又是怎样确定总共有多少任务、需要多少资源呢？接下来，我们就从一些重要的概念入手，对这些问题进行详细讲解。

4.3.1 数据流图

Flink是流式计算框架。它的程序结构，其实就是定义了一连串的处理操作，每个数据输入之后都会依次调用每步计算。在Flink代码中，我们定义的每个处理转换操作都叫作“算子”（Operator），所以我们的程序可以看作一串算子构成的管道，数据则像水流一样有序地流过。如在之前的WordCount代码中，基于执行环境调用的socketTextStream()方法，就是一个读取文本流的算子；而后面的flatMap()方法，则是将字符串数据进行分词、转换成二元组的算子。

所有的Flink程序都可以归纳为由3个部分构成：Source、Transformation和Sink。

· Source表示“源算子”，负责读取数据源。

· Transformation表示“转换算子”，利用各种算子进行处理加工。

· Sink表示“下沉算子”，负责数据的输出。

在运行时，Flink程序会被映射成所有算子按照逻辑顺序连接在一起的一张图，这被称为“逻辑数据流”（Logical Dataflow），或者叫“数据流图”（Dataflow Graph）。我们提交作业之后，打开Flink自带的Web UI，点击作业就能看到对应的数据流图，如图4-7所示。在数据流图中，可以清楚地看到Source、Transformation、Sink 3个部分。

数据流图类似于任意的有向无环图（DAG），这一点与Spark等其他框架是一致的。图中的每条数据流都以一个或多个Source算子开始，以一个或多个Sink算子结束。

在大部分情况下，数据流图中的算子和程序中的转换运算是一一对应的关系。那是不是说，我们代码中基于DataStream API的每个方法调用，都是一个算子呢？

并非如此。除了Source读取数据和Sink输出数据，一个中间的转换算子（Transformation Operator）必须是一个转换处理的操作；而在代码中有一些方法调用，数据是没有完成转换的，可能只是对属性做了一个设置，也可能定义的是数据的传递方式而非转换，又或者是需要几个方法合在一起才能表达一个完整的转换操作。例如，在之前的代码中，我们用到了定义分组的方法keyBy()，它就是一个数据分区操作，而并不是一个算子。事实上，在代码中我们可以看到调用其他转换操作之后返回的数据类型是SingleOutputStreamOperator，说明这是一个算子操作；而keyBy()返回的数据类型是KeyedStream。感兴趣的读者也可以自行在Web UI中提交任务并查看。

4.3.2 并行度

我们已经清楚了算子和数据流图的概念，那最终执行的任务又是什么呢？容易想到，一个算子操作就应该是一个任务。那是不是程序中的算子数量，就是最终执行的任务数呢？

1.什么是并行计算

要解答这个问题，我们需要先梳理一下其他框架分配任务、数据处理的过程。对Spark而言，是把根据程序生成的DAG划分阶段，进而分配任务的。对Flink这样的流式引擎，其实没有划分stage的必要，因为数据是连续不断到来的，我们完全可以按照数据流图建立一个“流水线”，前一个操作处理完成，就发往处理下一步操作的节点。如果说Spark基于MapReduce架构的思想是“数据不动代码动”，那么Flink就类似“代码不动数据流动”，原因就在于流式数据本身是连续到来的，我们不会同时传输所有数据，这其实是更符合数据流本身特点的处理方式。

在大数据场景下，我们依靠分布式架构进行并行计算，从而提高数据吞吐量的。既然处理完一个操作就可以把数据发往别处，那我们就可以将不同的算子操作任务，分配到不同的节点上执行了。这样就对任务做了分摊，实现了并行处理。

但是仔细分析会发现，这种“并行”其实并不彻底，因为算子之间是有执行顺序的，对一条数据来说必须依次执行；而一个算子在同一时刻只能处理一个数据。比如之前WordCount，一条数据到来之后，我们必须先用Source算子读进来、再做flatMap转换；一条数据被Source读入的同时，之前的数据可能正在被flatMap处理，这样不同的算子任务是并行的。但如果多条数据同时到来，一个算子是没有办法同时处理的，我们还是需要等待一条数据处理完，再处理下一条数据——这并没有真正提高吞吐量。

所以相对于上述的“任务并行”，我们真正关心的，是“数据并行”。也就是说，多条数据同时到来，我们应该可以同时读入，同时在不同节点执行flatMap操作。

2.并行子任务和并行度

怎样实现数据并行呢？其实也很简单，我们把一个算子操作，“复制”多份到多个节点，数据来了之后就可以到其中任意一个执行。这样一来，一个算子操作就被拆分成了多个并行的“子任务”，再将它们分发到不同节点，就真正实现了并行计算。

在Flink执行过程中，每个算子可以包含一个或多个子任务（operator subtask），这些子任务在不同的线程、不同的物理机或不同的容器中完全独立地执行。

一个特定算子的子任务的个数被称为并行度（parallelism）。这样，包含并行子任务的数据流，就是并行数据流，它需要多个分区（stream partition）来分配并行任务。在一般情况下，一个流程序的并行度，可以认为就是其所有算子中最大的并行度。在一个程序中，不同的算子可能具有不同的并行度。

如图4-8所示，当前数据流中有Source、map()、keyBy()/window()/apply()、Sink 4个算子，除了最后的Sink，其他算子的并行度都为2。整个程序包含了7个子任务，至少需要2个分区来并行执行。我们可以说，这段流处理程序的并行度就是2。

3.并行度的设置

在Flink中，可以用不同的方法来设置并行度，它们的有效范围和优先级别也是不同的。

（1）在代码中设置。

我们在代码中，可以很简单地在算子后跟着调用setParallelism()方法，来设置当前算子的并行度。

这种方式设置的并行度，只对当前算子有效。

另外，我们也可以直接调用执行环境的setParallelism()方法，全局设定并行度。

这样代码中的所有算子，默认的并行度就都为2了。我们一般不会在程序中设置全局并行度，因为如果在程序中对全局并行度进行硬编码，会导致无法动态扩容。

这里要注意的是，由于keyBy()方法返回的不是算子，所以无法对keyBy()设置并行度。

（2）在提交作业时设置。

在使用flink run命令提交作业时，可以增加-p参数来指定当前应用程序执行的并行度，它的作用类似于执行环境的全局设置。

如果我们直接在Web UI上提交作业，也可以在对应输入框中直接添加并行度。

（3）在配置文件中设置。

我们还可以直接在集群的配置文件flink-conf.yaml中更改默认并行度。

这个设置对于整个集群上提交的所有作业有效，初始值为1。无论在代码中设置，还是提交时的-p参数，都不是必须的。所以，在没有指定并行度的时候，就会采用配置文件中的集群默认并行度。在开发环境中，没有配置文件，默认并行度就是当前机器的CPU核心数。这也就解释了为什么我们在第2章运行WordCount流处理程序时，会看到结果前有1～4的分区编号——运行程序的计算机是4核CPU，那么开发环境默认的并行度就是4。

我们可以总结一下所有的并行度设置方法，它们的优先级如下所示。

（1）对于一个算子，首先看在代码中是否单独指定了它的并行度，这个特定的设置优先级最高，会覆盖后面所有的设置。

（2）如果没有单独设置，那么采用当前代码中执行环境全局设置的并行度。

（3）如果代码中完全没有设置，那么采用提交时-p参数指定的并行度。

（4）如果提交时也未指定-p参数，那么采用集群配置文件中的默认并行度。

这里需要说明的是，算子的并行度有时会受到自身具体实现的影响。比如之前我们用到的读取socket文本流的算子socketTextStream()，它本身就是非并行的Source算子，所以无论怎么设置，它在运行时的并行度都是1，对应在数据流图上就只有一个并行子任务。这一点大家可以自行在Web UI上查看验证。

那么实践中怎样设置并行度比较好呢？那就是在代码中只针对算子设置并行度，不设置全局并行度，方便我们在提交作业时进行动态扩容。

4.3.3 算子链

关于“一个作业有多少任务”这个问题，现在已经基本解决了。但如果我们仔细观察Web UI任务执行图，如图4-9所示，上面的节点似乎跟代码中的算子又不是一一对应的。

很明显，这里的一个节点，会把转换处理的很多个任务都连接在一起，合并成一个“大任务”。这又是怎么回事呢？

1.算子任务间的数据传输

回到上一节的例子，我们先来考察一下算子任务之间数据传输的方式。

如图4-8所示，一个数据流在算子之间传输数据的形式可以是一对一（one-to-one）的直通（forwarding）模式，也可以是打乱的重分区（redistributing）模式，具体是哪一种形式，取决于算子的种类。

（1）一对一的直通模式。

在这种模式下，数据流维护着分区及元素的顺序。如图4-8中的Source算子和map()算子，Source算子读取数据之后，可以直接发送给map()算子做处理，它们之间不需要重新分区，也不需要调整数据的顺序。这就意味着map()算子的子任务，看到的元素个数和顺序跟Source算子的子任务产生的完全一样，保证着“一对一”的关系。map()、filter()、flatMap()等算子都是这种one-to-one的对应关系。这种关系类似于Spark中的窄依赖。

（2）重分区模式。

在这种模式下，数据流的分区会发生改变。如图4-8中的map()和后面的keyBy()/window()/apply()算子之间[这里的keyBy()是数据传输方法，后面的window()、apply()方法共同构成了窗口算子]，以及窗口算子和Sink算子之间，都是这样的关系。

每个算子的子任务，都会根据数据传输的策略，把数据发送到不同的下游目标任务。例如，keyBy()是分组操作，本质上基于键（key）的哈希值（hashCode）进行了重分区；当并行度改变时，如从并行度为2的window算子，要传递到并行度为1的Sink算子，这时的数据传输方式是再平衡（rebalance），会把数据均匀地向下游子任务分发出去。这些传输方式都会引起重分区的过程，这一过程类似于Spark中的shuffle。

总体来说，这种算子间的关系类似于Spark中的宽依赖。

2.合并算子链

在Flink中，并行度相同的一对一（one-to-one）算子操作，可以直接链接在一起形成一个“大”的任务，这样原来的算子就成为真正任务里的一部分，如图4-10所示。每个任务都会被一个线程执行。这样的技术被称为“算子链”（Operator Chain）。

如图4-10中的例子，Source和map()之间满足了算子链的要求，所以可以直接合并在一起，形成了一个任务；因为并行度为2，所以合并后的任务也有两个并行子任务。这样，这个数据流图所表示的作业最终会有5个任务，由5个线程并行执行。

Flink为什么要有算子链的设计呢？这是因为将算子链接成任务是非常有效的优化，可以减少线程之间的切换和基于缓存区的数据交换，在减少时延的同时提升吞吐量。

Flink默认会按照算子链的原则进行链接合并，如果我们想要禁止合并或自行定义，也可以在代码中对算子做一些特定的设置。

4.3.4 作业图与执行图

至此，我们已经彻底了解了由代码生成任务的过程，现在来做个梳理总结。

由Flink程序直接映射成的数据流图，也被称为逻辑流图，因为它们表示的是计算逻辑的高级视图。到具体执行环节时，我们还要考虑并行子任务的分配、数据在任务间的传输，以及合并算子链的优化。为了说明最终应该怎样执行一个流处理程序，Flink需要将逻辑流图进行解析，转换为物理图。

在这个转换过程中，有几个不同的阶段，会生成不同层级的图，其中最重要的就是作业图和执行图。Flink中任务调度执行的图，按照生成顺序可以分成4层：

逻辑流图→作业图→执行图→物理图。

我们可以回忆一下之前处理socket文本流的StreamWordCount程序。

如果提交时设置并行度为2：

那么根据之前的分析，除了socketTextStream()是非并行的Source算子，它的并行度始终为1，其他算子的并行度都为2。

接下来，我们分析一下程序对应4层调度图的演变过程，如图4-11所示。

1.逻辑流图（StreamGraph）

这是根据用户通过DataStream API编写的代码生成的最初的DAG图，用来表示程序的拓扑结构。这一步一般在客户端完成。

我们可以看到，逻辑流图中的节点，完全对应着代码中的4步算子操作：源算子（socketTextStream()）→扁平映射算子（flatMap()）→分组聚合算子（keyBy()/sum()）→输出算子（print()）。

2.作业图（JobGraph）

StreamGraph经过优化后生成的就是作业图（JobGraph），这是提交给JobManager的数据结构，确定了当前作业中所有任务的划分。主要的优化为：将多个符合条件的节点链接在一起合并成一个任务节点，形成算子链，这样可以减少数据交换的消耗。JobGraph一般也是在客户端生成的，在作业提交时传递给JobMaster。

在图4-11中，分组聚合算子（Keyed Aggregation）和输出算子Sink（print）并行度都为2，而且是一对一的关系，满足算子链的要求，所以会合并在一起，成为一个任务节点。

3.执行图（ExecutionGraph）

JobMaster收到JobGraph后，会根据它来生成执行图（ExecutionGraph）。ExecutionGraph是JobGraph的并行化版本，是调度层最核心的数据结构。

从图4-11中可以看到，执行图与作业图最大的区别就是按照并行度对并行子任务进行了拆分，并明确了算子任务间数据传输的方式。

4.物理图（PhysicalGraph）

JobMaster生成执行图后，会将它分发给TaskManager；各个TaskManager会根据执行图来部署任务，最终的物理执行过程也会形成一张“图”，一般就叫作物理图（PhysicalGraph）。这只是具体执行层面的图，并不是一个具体的数据结构。

对应在图4-11中，物理图主要是在执行图的基础上，进一步确定数据存放的位置和收发的具体方式。有了物理图，TaskManager就可以对传递来的数据进行处理计算了。

我们可以看到，程序里定义了4个算子操作：源→转换→分组聚合→输出；在合并算子链进行优化之后，就只有3个任务节点了；再考虑并行度后，一共有5个并行子任务，最终需要5个线程来执行。

4.3.5 任务和任务槽

通过前几节的介绍，我们对任务的生成和分配已经非常清楚了。4.3.4节中我们最终得到结论：作业划分为5个并行子任务，需要5个线程并行执行。那在我们将应用提交到Flink集群之后，到底需要占用多少资源呢？是否需要5个TaskManager来运行呢？

1.任务槽

之前已经提到过，Flink中每个Worker（也就是TaskManager）都是一个JVM进程，它可以启动多个独立的线程，来并行执行多个子任务。

所以如果想要执行5个任务，并不一定非要5个TaskManager，我们可以让TaskManager多线程执行任务。如果可以同时运行5个线程，那么只要一个TaskManager就可以满足我们之前程序的运行需求了。

很显然，TaskManager的计算资源是有限的，并不是所有任务都可以放在一个TaskManager上并行执行。并行的任务越多，每个线程的资源就会越少。那一个TaskManager到底能并行处理多少个任务呢？为了控制并发量，我们需要在TaskManager上对每个任务运行所占用的资源都做出明确的划分，这就是所谓的任务槽。

slot的概念其实在分布式框架中并不陌生。所谓的“槽”是一种形象的表达，如果大家见过传说中的“卡带式游戏机”，就会对它有更直观的认识：游戏机上的卡槽提供了可以运行游戏的接口和资源，我们把游戏卡带插入卡槽，就可以占用游戏机的计算资源，执行卡带中的游戏程序了。一台经典的小霸王游戏机（见图4-12）一般只有一个卡槽，而在TaskManager中，我们可以设置多个slots，只要插入“卡带”——也就是分配好的任务，就可以并行执行了。

每个任务槽其实表示了TaskManager拥有计算资源的一个固定大小的子集，这些资源就是用来独立执行一个子任务的。

假如一个TaskManager有3个slots，那么它会将管理的内存平均分成3份，每个slot独自占据一份。这样一来，我们在slots上执行一个子任务时，相当于划定了一块内存“专款专用”，就不需要跟来自其他作业的任务去竞争内存资源了。因此现在我们只要2个TaskManager，就可以并行处理分配好的5个任务了，如图4-13所示。

2.任务槽数量的设置

我们可以通过集群的配置文件来设定TaskManager的slots数量：

通过调整slots的数量，我们就可以控制子任务之间的隔离级别了。

具体来说，如果一个TaskManager只有一个slot，那意味着每个任务都会运行在独立的JVM中（当然，该JVM可能是通过一个特定的容器启动的）；而一个TaskManager设置多个slots则意味着多个子任务可以共享同一个JVM。它们的区别在于：前者任务之间完全独立运行，隔离级别更高、彼此间的影响可以降到最小；而后者在同一个JVM进程中运行的任务将共享TCP连接和心跳消息，也可能共享数据集和数据结构，这就减少了每个任务的运行开销，在降低隔离级别的同时提升了性能。

需要注意的是，slots目前仅仅用来隔离内存，不会涉及CPU的隔离。在具体应用时，可以将slots数量配置为机器的CPU核心数，尽量避免不同任务之间对CPU的竞争。这也是开发环境默认并行度设为机器CPU核心数的原因。

3.任务对任务槽的共享

这样看来，一共有多少任务，我们就需要有多少slots来并行处理它们。但实际提交作业进行测试就会发现，我们之前的WordCount程序设置并行度为2提交，一共有5个并行子任务，可集群即使只有2个Task Slots也是可以成功提交并运行的。这又是为什么呢？

我们可以基于之前的例子继续扩展。如果我们保持Sink任务并行度为1不变，而作业提交时设置全局并行度为6，那么前两个任务节点就会各自有6个并行子任务，整个流处理程序则有13个子任务。那对于2个TaskManager、每个有3个slots的集群配置来说，能否正常运行呢？

完全没有问题。这是因为在默认情况下，Flink是允许子任务共享slots的。如图4-14所示，只要属于同一个作业，那么对于不同任务节点的并行子任务，就可以放到同一个slot上执行。对于第一个任务节点Source→map()，它的6个并行子任务必须分到不同的slots上（如果在同一个slot就没办法数据并行了），而第二个任务节点keyBy()/window()/apply()的并行子任务却可以和第一个任务节点共享slots。

于是最终结果就变成了：每个任务节点的并行子任务一字排开，占据不同的slots；而不同的任务节点的子任务可以共享slots。在一个slot中，可以将程序处理的所有任务都放在这里执行，我们把它叫作保存了整个作业的运行管道（pipeline）。

这个特性看起来有点奇怪：我们不是希望并行处理、任务之间相互隔离吗，为什么这里又允许共享slots呢？

我们知道，一个slot对应了一组独立的计算资源。在之前不做共享的时候，每个任务都平等地占据了一个slot，但其实不同的任务对资源的占用是不同的。例如，这里的前两个任务，Source/map()尽管是两个算子合并算子链得到的，但它只是基本的数据读取和简单转换，计算耗时极短，一般也不需要太大的内存空间；而window算子所做的窗口操作，往往会涉及大量的数据、状态存储和计算，我们一般把这类任务叫作“资源密集型”（intensive）任务。当它们被平等地分配到独立的slot上时，实际运行中我们就会发现，大量数据到来时Source/map()和Sink任务很快就可以完成，但window任务却耗时很久；于是下游的Sink任务占据的slots就会等待闲置，而上游的Source/map()任务受限于下游的处理能力，也会在快速处理完一部分数据后阻塞对应的资源开始等待（相当于处理背压）。这样资源的利用就出现了极大的不平衡，“忙的忙死，闲的闲死”。

解决这一问题的思路就是允许slots共享。当我们将资源密集型和非密集型的任务同时放到一个slot中时，它们就可以自行分配对资源占用的比例，从而保证最重的活平均分配给所有的TaskManager。

slot共享的另一个好处就是允许我们保存完整的作业管道。这样一来，即使某个TaskManager出现故障宕机，其他节点也可以完全不受影响，作业的任务可以继续执行。

另外，同一个任务节点的并行子任务是不能共享slots的，所以允许slots共享之后，运行作业所需的slots数量正好就是作业中所有算子并行度的最大值。这样一来，我们考虑当前集群需要配置多少slots资源时，就不需要再去详细计算一个作业总共包含多少个并行子任务了，只看最大的并行度就够了。

当然，Flink默认是允许slots共享的，如果希望某个算子对应的任务完全独占一个slot，或者只有某一部分算子共享slots，我们也可以通过设置“slot共享组”（SlotSharingGroup）手动指定。

这样，只有属于同一个slot共享组的子任务，才会开启slots共享；不同组之间的任务是完全隔离的，必须分配到不同的slots上。在这种场景下，总共需要的slots数量，就是各个slots共享组最大并行度的总和。

4.任务槽和并行度的关系

直观上看，slots就是TaskManager为了并行执行任务而设置的，那它和之前讲过的并行度（Parallelism）是不是一回事呢？

slots和并行度确实都跟程序的并行执行有关，但两者是完全不同的概念。简单来说，Task Slots是静态的概念，是指TaskManager具有的并发执行能力，可以通过参数taskmanager.numberOfTaskSlots进行配置；而并行度是动态概念，也就是TaskManager运行程序时实际使用的并发能力，可以通过参数parallelism.default进行配置。换句话说，并行度如果小于或等于集群中可用slots的总数，程序是可以正常执行的，因为slots不一定要全部占用，有十分力气可以只用八分；而如果并行度大于可用slots总数，导致超出了并行能力上限，那么心有余力不足，程序就只好等待资源管理器分配更多的资源了。

下面我们再举一个具体的例子。假设一共有3个TaskManager，每个TaskManager中的slots数量设置为3个，那么一共有9个Slots，如图4-15所示，表示集群最多能并行执行9个任务。

而我们定义WordCount程序的处理操作是4个转换算子：Source→FlatMap→Reduce→Sink。

当所有算子并行度相同时，容易看出Source和FlatMap可以合并算子链，于是最终有3个任务节点。

如果我们没有任何并行度设置，而配置文件中默认parallelism.default=1，那么程序运行的默认并行度为1，总共有3个任务。由于不同算子的任务可以共享任务槽，所以最终占用的slot只有1个。9个slots只用了1个，有8个空闲，如图4-16中的示例所示。

如果我们更改默认参数，或者提交作业时设置并行度为2，那么总共有6个任务，共享任务槽之后会占用2个slots，如图4-17中示例2所示。同样，就有7个slots空闲，计算资源没有被充分利用。所以可以看到，设置合适的并行度才能提高效率。

那对于这个例子，怎样设置并行度效率最高呢？当然是把所有的slots都利用起来。考虑到slots共享，我们可以直接把并行度设置为9，这样所有27个任务就会完全占用9个slots。这是当前集群资源下能执行的最大并行度，计算资源得到了充分的利用，如图4-18中示例3所示。

另外再考虑对某个算子单独设置并行度的场景。例如，如果我们考虑到输出可能是写入文件，那会希望不要并行写入多个文件，就需要设置Sink算子的并行度为1。这时其他的算子并行度依然为9，所以总共会有19个子任务。根据slots共享的原则，它们最终还是会占用全部的9个slots，而Sink任务只在其中一个slot上执行，如图4-19中示例4所示。通过这个例子也可以明确地看到，整个流处理程序的并行度，就应该是所有算子并行度中最大的那个，这代表了运行程序需要的slots数量。