Flink Runtime详解

Flink运行时架构

Flink集群的架构，也是典型的主从架构模型，启动Flink，它会启动JobManager以及TaskManager进程，这里以standalone模式来讲解其运行时的架构以及一些细节，集群模式后面的文章汇介绍到。

JobManager是master主节点，它主要负责资源的分配、任务的调度以及集群的管理。TaskManager是从节点，如果单独运行在一台机器上则可以称作一个slave节点。主从节点之间通过心跳机制保持联系，当我们从client提交一个job的时候，会将job提交到jobmanager主节点，然后由主节点分配到各个slave节点。各个节点之间消息的通信使用akka框架来完成，数据的传输基于Netty框架来完成。整个架构如下图所示：

提交的job会分配给集群中的slave节点去执行，即由每个节点上的TaskManager来执行，在每一个TaskManager中，所有的资源会被进行划分（这里指内存等资源，但不包括CPU），被划分为多个slot，整个job被划分为多个task，被分配到这些slot中执行，每个slot对应与一个task，但这并不是说一个slot只能运行一个task，slot的数量代表task的并行度，因为我们可以通过slot共享的方式让多个task共享一个slot，从而充分利用集群的资源。

一个task是提交的job中的一个或者多个operator的组合，如果是多个operator组合而成，则也可以成为一个operatorChain。在同一个TaskManager中的task会共享TCP连接（通过多路复用）以及心跳消息以减小开销。

为了提高更加高效的执行提交的job，Flink会尽可能地将多个operator连接成一个operatorChain。以wordcount为例，下图中将KeyAggregation和sink两个operator合并，这是因为这两个操作合并之后并不会改变整个拓扑结构。

要想将多个operator连接成一个operatorChain，需要满足以下几点条件：

1. 没有禁用chain
2. 上下游算子并行度一致
3. 下游 算子的入度为1（即下游节点没有来自其他节点的输入）。
4. 上下游算子在一个slot group中，可以手动指定。
5. 下游节点的chain的策略为Always，即可以与上下游连接（map，flatmap，filter算子的chain策略默认为Always）。
6. 上游节点的的chain策略为Always或者HEAD（即只能与下游连接，不能与上游连接，Source默认为HEAD）。
7. 上下游节点之间没有Shuffle操作，即为Forward。

使用operatorChain之后，会有如下的优点：

1. 减少线程之间切换（因为如果不连接chain，那么会单独成为一个task，也就是启动一个线程执行）。
2. 减少序列化以及反序列化。
3. 减少数据在缓冲区之间的交换。
4. 减少延迟并且提高吞吐能力。

我们也可以在程序中手动指定OperatorChain：

• 可以通过在DataStream的operator后面（如someStream.map(..))调用startNewChain()来指示从该operator开始一个新的chain（与前面截断，不会被chain到前面）。
• 调用disableChaining()来指示该operator不参与chaining，即不与前后的chain连接在一起。
• 通过调用StreamExecutionEnvironment.disableOperatorChaining()来全局禁用chaining。
• 设置Slot group，例如someStream.filter(...).slotSharingGroup(“name”)。
• 调整并行度（若要在一个chain中必须并行度相同）。

上面我们提到多在TaskManager中会划分为多个slot，在slot中执行task，这个task是由多个operator连接在一起的chain组成的。我们可以通过slot共享的方式来充分利用集群的资源。

在默认的情况下，Flink执行subtask共享slot，条件是这些subtask必须是来自一个job的task的subtask，在有些情况下，有些一个slot可能持有一个job的整个pipeline流程。共享slot有以下节点好处：

1. Flink中的slot数量和job中的最大并行度一致，这样就不需要去计算task的数量。
2. 充分利用集群的资源，如果不共享slot则可能非密集型操作会和密集型操作占用相同的资源，共享slot之后，充分利用了资源，可以使得每个TaskManager的每个slot数量较少，节约资源，并且每个slot具有相同的负载。

在Flink中，使用SlotSharingGroup类来实现slot之间的共享，尽可能的让多个subtask共享一个slot。保证一个job下并行度相同的subtask共享一个slot，算子默认的group为default（即默认共享slot）。

有些情况下，我们并不需要一些subtask共享slot，所以会使用下面的方式来为一个算子重新指定一个group。someStream.filter(...).slotSharingGroup("group1")，即filter算子重新 指定了group1，即不与前后操作在同一个chain中。在一些场景下，适当的指定group可以减少单个slot中的负载。

接着我们说下task slot的数量和task并行度的关系，其中每个task中的slot的数量和task中的最高并行度相同。

那么在整个job中，所需的slot数量是每个task的最大并行度之和。如下图所示，需要的slot数量为30。

在Flink中的transformation之间，存在多种分区策略，如果每个操作的并行度为1，那么对应的着一对一的分区策略，另外还有几种不同的分区策略。

• Random partitioning：按均匀分布随机划分元素，网络开销往往比较大dataStream.shuffle()。
• Round-robin partitioning：循环对元素进行分区，为每一个分区创建相等的负载，这在数据倾斜时非常有用dataStream.rebalance()。
• Rescaling：跟rebalance有点类似，但不是全局的，通过轮询调度将元素从上游的task一个子集发送到下游task的一个子集：dataStream.rescale()。
• Broadcasting：将元素广播到每个分区上dataStream.broadcast()。
• Custom partitioning：自定义分区方式dataStream.partitionCustom(partitioner, "someKey")。

需要注意区分Round-robin和Rescaling策略，前者是全局划分的，后者是针对单个TaskManager来划分的，这样可以避免一些网络连接和开销，提高效率。如下图所示，只针对同一个TaskManager来划分，而不是全局划分。